November 2009 - hornetq-commits - Jboss List Archives

JBoss hornetq SVN: r8191 - trunk/src/main/org/hornetq/core/persistence/impl/journal.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-03 04:23:31 -0500 (Tue, 03 Nov 2009) New Revision: 8191 Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java Log: fixed bug in large messages Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-11-03 08:48:47 UTC (rev 8190) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-11-03 09:23:31 UTC (rev 8191) @@ -66,7 +66,7 @@ public JournalLargeServerMessage(final JournalStorageManager storageManager) { - this.storageManager = storageManager; + this.storageManager = storageManager; } /** @@ -83,7 +83,7 @@ storageManager = copy.storageManager; file = fileCopy; bodySize = copy.bodySize; - setMessageID(newID); + setMessageID(newID); } // Public -------------------------------------------------------- @@ -259,7 +259,7 @@ public boolean isFileExists() throws Exception { - SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID()); + SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID(), durable); return localfile.exists(); } @@ -305,7 +305,7 @@ idToUse = linkMessage.getMessageID(); } - SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse); + SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse, durable); ServerMessage newMessage = new JournalLargeServerMessage(linkMessage == null ? this : (JournalLargeServerMessage)linkMessage, @@ -341,9 +341,9 @@ { throw new RuntimeException("MessageID not set on LargeMessage"); } + + file = storageManager.createFileForLargeMessage(getMessageID(), durable); - file = storageManager.createFileForLargeMessage(getMessageID()); - file.open(); bodySize = file.size(); @@ -364,7 +364,7 @@ this.linkMessage = message; - file = storageManager.createFileForLargeMessage(message.getMessageID()); + file = storageManager.createFileForLargeMessage(message.getMessageID(), durable); try { file.open(); Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-03 08:48:47 UTC (rev 8190) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-03 09:23:31 UTC (rev 8191) @@ -946,6 +946,7 @@ { // this could happen if the message was deleted but the file still exists as the file still being used originalMessage = createLargeMessage(); + originalMessage.setDurable(true); originalMessage.setMessageID(originalMessageID); messages.put(originalMessageID, originalMessage); } @@ -1372,9 +1373,16 @@ * @param messageID * @return */ - SequentialFile createFileForLargeMessage(final long messageID) + SequentialFile createFileForLargeMessage(final long messageID, final boolean durable) { - return largeMessagesFactory.createSequentialFile(messageID + ".msg", -1); + if (durable) + { + return largeMessagesFactory.createSequentialFile(messageID + ".msg", -1); + } + else + { + return largeMessagesFactory.createSequentialFile(messageID + ".tmp", -1); + } } // Private ----------------------------------------------------------------------------------

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8190 - trunk/tests/src/org/hornetq/tests/integration/cluster/distribution.

by do-not-reply＠jboss.org

Author: ataylor Date: 2009-11-03 03:48:47 -0500 (Tue, 03 Nov 2009) New Revision: 8190 Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java Log: fixed tests Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java 2009-11-03 08:45:42 UTC (rev 8189) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusteredGroupingTest.java 2009-11-03 08:48:47 UTC (rev 8190) @@ -193,6 +193,7 @@ } } + public void testGroupingSendTo2queues() throws Exception { setupServer(0, isFileStorage(), isNetty()); @@ -230,9 +231,9 @@ waitForBindings(1, "queues.testaddress", 1, 1, true); waitForBindings(2, "queues.testaddress", 1, 1, true); - /*waitForBindings(0, "queues.testaddress", 2, 2, false); + waitForBindings(0, "queues.testaddress", 2, 2, false); waitForBindings(1, "queues.testaddress", 2, 2, false); - waitForBindings(2, "queues.testaddress", 2, 2, false);*/ + waitForBindings(2, "queues.testaddress", 2, 2, false); sendInRange(0, "queues.testaddress", 0, 10, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); @@ -796,76 +797,7 @@ } } - public void testGroupingSendTo3queuesSendingNodeGoesDown() throws Exception - { - setupServer(0, isFileStorage(), isNetty()); - setupServer(1, isFileStorage(), isNetty()); - setupServer(2, isFileStorage(), isNetty()); - setupClusterConnection("cluster0", "queues", false, 1, isNetty(), 0, 1, 2); - - setupClusterConnection("cluster1", "queues", false, 1, isNetty(), 1, 0, 2); - - setupClusterConnection("cluster2", "queues", false, 1, isNetty(), 2, 0, 1); - - - setUpGroupHandler(GroupingHandlerConfiguration.TYPE.LOCAL, 0); - setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 1); - setUpGroupHandler(GroupingHandlerConfiguration.TYPE.REMOTE, 2); - - startServers(0, 1, 2); - - try - { - - setupSessionFactory(0, isNetty()); - setupSessionFactory(1, isNetty()); - setupSessionFactory(2, isNetty()); - - createQueue(0, "queues.testaddress", "queue0", null, true); - createQueue(1, "queues.testaddress", "queue0", null, true); - createQueue(2, "queues.testaddress", "queue0", null, true); - - addConsumer(0, 1, "queue0", null); - - waitForBindings(0, "queues.testaddress", 1, 0, true); - waitForBindings(1, "queues.testaddress", 1, 1, true); - waitForBindings(2, "queues.testaddress", 1, 0, true); - - waitForBindings(0, "queues.testaddress", 2, 1, false); - waitForBindings(1, "queues.testaddress", 2, 0, false); - waitForBindings(2, "queues.testaddress", 2, 1, false); - - sendInRange(1, "queues.testaddress", 0, 10, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); - - verifyReceiveAllInRange(0, 10, 0); - closeSessionFactory(0); - stopServers(0); - - sendInRange(2, "queues.testaddress", 10, 20, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); - - - setUpGroupHandler(GroupingHandlerConfiguration.TYPE.LOCAL, 0); - startServers(0); - waitForBindings(0, "queues.testaddress", 1, 0, true); - setupSessionFactory(0, isNetty()); - verifyReceiveAllInRange(10, 20, 0); - sendInRange(0, "queues.testaddress", 20, 30, false, MessageImpl.HDR_GROUP_ID, new SimpleString("id1")); - - verifyReceiveAllInRange(20, 30, 0); - - System.out.println("*****************************************************************************"); - } - finally - { - //closeAllConsumers(); - - closeAllSessionFactories(); - - stopServers(0, 1, 2); - } - } - public void testGroupingMultipleQueuesOnAddress() throws Exception { setupServer(0, isFileStorage(), isNetty());

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8189 - in trunk: src/main/org/hornetq/core/postoffice/impl and 9 other directories.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-03 03:45:42 -0500 (Tue, 03 Nov 2009) New Revision: 8189 Removed: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCleanupTest.java trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationLargemessageEndMessage.java trunk/src/main/org/hornetq/core/replication/ReplicationManager.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java trunk/src/main/org/hornetq/core/server/ServerMessage.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/core/transaction/impl/TransactionImpl.java trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java Log: removed over-complex large message cleanup code Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalLargeServerMessage.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -36,6 +36,9 @@ * * */ + + +//FIXME - this class should be renamed to just large message public class JournalLargeServerMessage extends ServerMessageImpl implements LargeServerMessage { // Constants ----------------------------------------------------- @@ -54,6 +57,8 @@ private SequentialFile file; private long bodySize = -1; + + private final AtomicInteger delayDeletionCount = new AtomicInteger(0); // Static -------------------------------------------------------- @@ -177,29 +182,18 @@ public void decode(final HornetQBuffer buffer) { file = null; - try - { - this.setStored(); - } - catch (Exception e) - { - // File still null, this wasn't supposed to happen ever. - log.warn(e.getMessage(), e); - } decodeHeadersAndProperties(buffer); } - private final AtomicInteger delayDeletionCount = new AtomicInteger(0); - public synchronized void incrementDelayDeletionCount() { this.delayDeletionCount.incrementAndGet(); } - + public synchronized void decrementDelayDeletionCount() throws Exception { int count = this.delayDeletionCount.decrementAndGet(); - + if (count == 0) { checkDelete(); @@ -207,7 +201,7 @@ } private void checkDelete() throws Exception - { + { if (getRefCount() <= 0) { if (linkMessage != null) @@ -265,7 +259,7 @@ public boolean isFileExists() throws Exception { - SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID(), isStored()); + SequentialFile localfile = storageManager.createFileForLargeMessage(getMessageID()); return localfile.exists(); } @@ -284,18 +278,6 @@ return memoryEstimate; } - @Override - public void setStored() throws Exception - { - super.setStored(); - releaseResources(); - - if (file != null && linkMessage == null) - { - storageManager.completeLargeMessage(this); - } - } - public synchronized void releaseResources() { if (file != null && file.isOpen()) @@ -323,12 +305,12 @@ idToUse = linkMessage.getMessageID(); } - SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse, isStored()); + SequentialFile newfile = storageManager.createFileForLargeMessage(idToUse); ServerMessage newMessage = new JournalLargeServerMessage(linkMessage == null ? this - : (JournalLargeServerMessage)linkMessage, - newfile, - newID); + : (JournalLargeServerMessage)linkMessage, + newfile, + newID); return newMessage; } @@ -360,7 +342,7 @@ throw new RuntimeException("MessageID not set on LargeMessage"); } - file = storageManager.createFileForLargeMessage(getMessageID(), isStored()); + file = storageManager.createFileForLargeMessage(getMessageID()); file.open(); @@ -369,14 +351,6 @@ } } -// /* (non-Javadoc) -// * @see org.hornetq.core.server.LargeServerMessage#getLinkedMessage() -// */ -// public LargeServerMessage getLinkedMessage() -// { -// return linkMessage; -// } - /* (non-Javadoc) * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) */ @@ -390,7 +364,7 @@ this.linkMessage = message; - file = storageManager.createFileForLargeMessage(message.getMessageID(), true); + file = storageManager.createFileForLargeMessage(message.getMessageID()); try { file.open(); Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -95,8 +95,9 @@ private static final long CHECKPOINT_BATCH_SIZE = Integer.MAX_VALUE; - //grouping journal record type + // grouping journal record type public static final byte GROUP_RECORD = 41; + // Bindings journal record type public static final byte QUEUE_BINDING_RECORD = 21; @@ -190,13 +191,13 @@ SequentialFileFactory bindingsFF = new NIOSequentialFileFactory(bindingsDir); Journal localBindings = new JournalImpl(1024 * 1024, - 2, - config.getJournalCompactMinFiles(), - config.getJournalCompactPercentage(), - bindingsFF, - "hornetq-bindings", - "bindings", - 1); + 2, + config.getJournalCompactMinFiles(), + config.getJournalCompactPercentage(), + bindingsFF, + "hornetq-bindings", + "bindings", + 1); if (replicator != null) { @@ -254,17 +255,17 @@ } else { - this.idGenerator = new BatchingIDGenerator(0, CHECKPOINT_BATCH_SIZE, bindingsJournal); + this.idGenerator = new BatchingIDGenerator(0, CHECKPOINT_BATCH_SIZE, bindingsJournal); } Journal localMessage = new JournalImpl(config.getJournalFileSize(), - config.getJournalMinFiles(), - config.getJournalCompactMinFiles(), - config.getJournalCompactPercentage(), - journalFF, - "hornetq-data", - "hq", - config.getJournalMaxAIO()); + config.getJournalMinFiles(), + config.getJournalCompactMinFiles(), + config.getJournalCompactPercentage(), + journalFF, + "hornetq-data", + "hq", + config.getJournalMaxAIO()); if (replicator != null) { @@ -471,8 +472,8 @@ public void updateScheduledDeliveryTime(final MessageReference ref) throws Exception { - ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), - ref.getQueue().getID()); + ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), ref.getQueue() + .getID()); messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(), SET_SCHEDULED_DELIVERY_TIME, @@ -548,12 +549,12 @@ messageJournal.appendAddRecord(id, HEURISTIC_COMPLETION, new HeuristicCompletionEncoding(xid, isCommit), true); return id; } - + public void deleteHeuristicCompletion(long id) throws Exception { messageJournal.appendDeleteRecord(id, true); } - + public void deletePageTransactional(final long txID, final long recordID) throws Exception { messageJournal.appendDeleteRecordTransactional(txID, recordID); @@ -561,8 +562,8 @@ public void updateScheduledDeliveryTimeTransactional(final long txID, final MessageReference ref) throws Exception { - ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), - ref.getQueue().getID()); + ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), ref.getQueue() + .getID()); messageJournal.appendUpdateRecordTransactional(txID, ref.getMessage().getMessageID(), @@ -627,19 +628,6 @@ updateInfo, syncNonTransactional); } - /** - * @param journalLargeServerMessage - * @throws Exception - */ - public void completeLargeMessage(JournalLargeServerMessage message) throws Exception - { - if (isReplicated()) - { - replicator.largeMessageEnd(message.getMessageID()); - } - SequentialFile fileToRename = createFileForLargeMessage(message.getMessageID(), true); - message.getFile().renameTo(fileToRename.getFileName()); - } private static final class AddMessageRecord { @@ -654,11 +642,11 @@ int deliveryCount; } - + private class LargeMessageTXFailureCallback implements TransactionFailureCallback { private final Map<Long, ServerMessage> messages; - + public LargeMessageTXFailureCallback(Map<Long, ServerMessage> messages) { super(); @@ -687,7 +675,7 @@ } } } - + } public void loadMessageJournal(final PostOffice postOffice, @@ -701,7 +689,7 @@ List<PreparedTransactionInfo> preparedTransactions = new ArrayList<PreparedTransactionInfo>(); Map<Long, ServerMessage> messages = new HashMap<Long, ServerMessage>(); - + messageJournal.load(records, preparedTransactions, new LargeMessageTXFailureCallback(messages)); ArrayList<LargeServerMessage> largeMessages = new ArrayList<LargeServerMessage>(); @@ -926,7 +914,7 @@ msg.decrementDelayDeletionCount(); } } - + if (perfBlastPages != -1) { messageJournal.perfBlast(perfBlastPages); @@ -946,25 +934,24 @@ LargeMessageEncoding messageEncoding = new LargeMessageEncoding(largeMessage); messageEncoding.decode(buff); - + Long originalMessageID = (Long)largeMessage.getProperties().getProperty(MessageImpl.HDR_ORIG_MESSAGE_ID); - + // Using the linked file by the original file if (originalMessageID != null) { LargeServerMessage originalMessage = (LargeServerMessage)messages.get(originalMessageID); - + if (originalMessage == null) { // this could happen if the message was deleted but the file still exists as the file still being used originalMessage = createLargeMessage(); originalMessage.setMessageID(originalMessageID); - originalMessage.setStored(); messages.put(originalMessageID, originalMessage); } - + originalMessage.incrementDelayDeletionCount(); - + largeMessage.setLinkedMessage(originalMessage); } return largeMessage; @@ -992,7 +979,7 @@ // Use same method as load message journal to prune out acks, so they don't get added. // Then have reacknowledge(tx) methods on queue, which needs to add the page size - + // first get any sent messages for this tx and recreate for (RecordInfo record : preparedTransaction.records) { @@ -1001,13 +988,13 @@ HornetQBuffer buff = ChannelBuffers.wrappedBuffer(data); byte recordType = record.getUserRecordType(); - + switch (recordType) { case ADD_LARGE_MESSAGE: - { + { messages.put(record.id, parseLargeMessage(messages, buff)); - + break; } case ADD_MESSAGE: @@ -1022,6 +1009,7 @@ } case ADD_REF: { + long messageID = record.id; RefEncoding encoding = new RefEncoding(); @@ -1163,11 +1151,13 @@ resourceManager.putTransaction(xid, tx); } } - - //grouping handler operations + + // grouping handler operations public void addGrouping(GroupBinding groupBinding) throws Exception { - GroupingEncoding groupingEncoding = new GroupingEncoding(groupBinding.getId(), groupBinding.getGroupId(), groupBinding.getClusterName()); + GroupingEncoding groupingEncoding = new GroupingEncoding(groupBinding.getId(), + groupBinding.getGroupId(), + groupBinding.getClusterName()); bindingsJournal.appendAddRecord(groupBinding.getId(), GROUP_RECORD, groupingEncoding, true); } @@ -1222,7 +1212,7 @@ bindingEncoding.setId(id); - queueBindingInfos.add(bindingEncoding); + queueBindingInfos.add(bindingEncoding); } else if (rec == PERSISTENT_ID_RECORD) { @@ -1236,7 +1226,7 @@ { idGenerator.loadState(record.id, buffer); } - else if(rec == GROUP_RECORD) + else if (rec == GROUP_RECORD) { GroupingEncoding encoding = new GroupingEncoding(); encoding.decode(buffer); @@ -1285,7 +1275,7 @@ // Must call close to make sure last id is persisted if (idGenerator != null) { - idGenerator.close(); + idGenerator.close(); } bindingsJournal.stop(); @@ -1371,7 +1361,7 @@ if (executor == null) { deleteAction.run(); - } + } else { executor.execute(deleteAction); @@ -1382,16 +1372,9 @@ * @param messageID * @return */ - SequentialFile createFileForLargeMessage(final long messageID, final boolean stored) + SequentialFile createFileForLargeMessage(final long messageID) { - if (stored) - { - return largeMessagesFactory.createSequentialFile(messageID + ".msg", -1); - } - else - { - return largeMessagesFactory.createSequentialFile(messageID + ".tmp", -1); - } + return largeMessagesFactory.createSequentialFile(messageID + ".msg", -1); } // Private ---------------------------------------------------------------------------------- @@ -1464,13 +1447,13 @@ return XidCodecSupport.getXidEncodeLength(xid); } } - + private static class HeuristicCompletionEncoding implements EncodingSupport { Xid xid; boolean isCommit; - + HeuristicCompletionEncoding(final Xid xid, final boolean isCommit) { this.xid = xid; @@ -1921,5 +1904,4 @@ } - } Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -600,8 +600,6 @@ { if (pagingManager.page(message, true)) { - message.setStored(); - return; } } @@ -668,7 +666,7 @@ } public MessageReference reroute(final ServerMessage message, final Queue queue, final Transaction tx) throws Exception - { + { MessageReference reference = message.createReference(queue); Long scheduledDeliveryTime = (Long)message.getProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME); @@ -680,8 +678,6 @@ message.incrementDurableRefCount(); - message.setStored(); - PagingStore store = pagingManager.getPageStore(message.getDestination()); message.incrementRefCount(store, reference); @@ -865,11 +861,11 @@ { reference.setScheduledDeliveryTime(scheduledDeliveryTime); } - + if (message.isDurable() && queue.isDurable()) { int durableRefCount = message.incrementDurableRefCount(); - + if (durableRefCount == 1) { if (tx != null) @@ -880,8 +876,6 @@ { storageManager.storeMessage(message); } - - message.setStored(); } if (tx != null) @@ -1163,7 +1157,6 @@ { if (pagingManager.page(message, tx.getID(), first)) { - message.setStored(); if (message.isDurable()) { // We only create pageTransactions if using persistent messages @@ -1232,7 +1225,7 @@ public void beforeRollback(Transaction tx) throws Exception { // Reverse the ref counts, and paging sizes - + for (MessageReference ref : refs) { ServerMessage message = ref.getMessage(); Modified: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationLargemessageEndMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationLargemessageEndMessage.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationLargemessageEndMessage.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -31,8 +31,6 @@ long messageId; - boolean isDelete; - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- @@ -42,11 +40,10 @@ super(REPLICATION_LARGE_MESSAGE_END); } - public ReplicationLargemessageEndMessage(final long messageId, final boolean isDelete) + public ReplicationLargemessageEndMessage(final long messageId) { this(); this.messageId = messageId; - this.isDelete = isDelete; } // Public -------------------------------------------------------- @@ -54,21 +51,19 @@ @Override public int getRequiredBufferSize() { - return BASIC_PACKET_SIZE + DataConstants.SIZE_LONG + DataConstants.SIZE_BOOLEAN; + return BASIC_PACKET_SIZE + DataConstants.SIZE_LONG; } @Override public void encodeBody(final HornetQBuffer buffer) { buffer.writeLong(messageId); - buffer.writeBoolean(isDelete); } @Override public void decodeBody(final HornetQBuffer buffer) { messageId = buffer.readLong(); - isDelete = buffer.readBoolean(); } /** @@ -79,14 +74,6 @@ return messageId; } - /** - * @return the isDelete - */ - public boolean isDelete() - { - return isDelete; - } - // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/replication/ReplicationManager.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -78,8 +78,6 @@ void largeMessageWrite(long messageId, byte [] body); - void largeMessageEnd(long messageId); - void largeMessageDelete(long messageId); } Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -66,7 +66,7 @@ // Attributes ---------------------------------------------------- private static final boolean trace = log.isTraceEnabled(); - + private static void trace(String msg) { log.trace(msg); @@ -85,7 +85,7 @@ private PagingManager pageManager; private final ConcurrentMap<SimpleString, ConcurrentMap<Integer, Page>> pageIndex = new ConcurrentHashMap<SimpleString, ConcurrentMap<Integer, Page>>(); - + private final ConcurrentMap<Long, LargeServerMessage> largeMessages = new ConcurrentHashMap<Long, LargeServerMessage>(); // Constructors -------------------------------------------------- @@ -201,7 +201,7 @@ { channel.close(); storage.stop(); - + for (ConcurrentMap<Integer, Page> map : pageIndex.values()) { for (Page page : map.values()) @@ -216,15 +216,14 @@ } } } - + pageIndex.clear(); - - + for (LargeServerMessage largeMessage : largeMessages.values()) { largeMessage.releaseResources(); } - + largeMessages.clear(); } @@ -254,30 +253,17 @@ */ private void handleLargeMessageEnd(ReplicationLargemessageEndMessage packet) { - LargeServerMessage message = lookupLargeMessage(packet.getMessageId(), packet.isDelete()); + LargeServerMessage message = lookupLargeMessage(packet.getMessageId(), true); + if (message != null) { - if (packet.isDelete()) + try { - try - { - message.deleteFile(); - } - catch (Exception e) - { - log.warn("Error deleting large message ID = " + packet.getMessageId(), e); - } + message.deleteFile(); } - else + catch (Exception e) { - try - { - message.setStored(); - } - catch (Exception e) - { - log.warn("Error deleting large message ID = " + packet.getMessageId(), e); - } + log.warn("Error deleting large message ID = " + packet.getMessageId(), e); } } } @@ -293,13 +279,12 @@ message.addBytes(packet.getBody()); } } - - - private LargeServerMessage lookupLargeMessage(long messageId, boolean isDelete) + + private LargeServerMessage lookupLargeMessage(long messageId, boolean delete) { LargeServerMessage message; - - if (isDelete) + + if (delete) { message = largeMessages.remove(messageId); } @@ -307,12 +292,12 @@ { message = largeMessages.get(messageId); } - + if (message == null) { log.warn("Large MessageID " + messageId + " is not available on backup server. Ignoring replication message"); } - + return message; } @@ -328,7 +313,6 @@ this.largeMessages.put(largeMessage.getMessageID(), largeMessage); } - /** * @param packet */ @@ -433,12 +417,11 @@ ConcurrentMap<Integer, Page> pages = getPageMap(packet.getStoreName()); Page page = pages.remove(packet.getPageNumber()); - + if (page == null) { page = getPage(packet.getStoreName(), packet.getPageNumber()); } - if (page != null) { Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -276,22 +276,11 @@ { if (enabled) { - sendReplicatePacket(new ReplicationLargemessageEndMessage(messageId, true)); + sendReplicatePacket(new ReplicationLargemessageEndMessage(messageId)); } } /* (non-Javadoc) - * @see org.hornetq.core.replication.ReplicationManager#largeMessageEnd(long) - */ - public void largeMessageEnd(long messageId) - { - if (enabled) - { - sendReplicatePacket(new ReplicationLargemessageEndMessage(messageId, false)); - } - } - - /* (non-Javadoc) * @see org.hornetq.core.replication.ReplicationManager#largeMessageWrite(long, byte[]) */ public void largeMessageWrite(long messageId, byte[] body) Modified: trunk/src/main/org/hornetq/core/server/ServerMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerMessage.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/server/ServerMessage.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -47,11 +47,6 @@ int getMemoryEstimate(); - //TODO - do we really need this? Can't we use durable ref count? - void setStored() throws Exception; - - boolean isStored(); - int getRefCount(); ServerMessage makeCopyForExpiryOrDLA(long newID, boolean expiry) throws Exception; Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -1379,9 +1379,10 @@ config.getName(), config.getAddress(), config.getTimeout()); - } - log.info("deploying grouping handler: " + groupingHandler); + } + this.groupingHandler = groupingHandler; + managementService.addNotificationListener(groupingHandler); } } Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -42,8 +42,6 @@ /** Global reference counts for paging control */ private final AtomicInteger refCount = new AtomicInteger(0); - private volatile boolean stored; - // We cache this private volatile int memoryEstimate = -1; @@ -102,16 +100,6 @@ return ref; } - public boolean isStored() - { - return stored; - } - - public void setStored() throws Exception - { - stored = true; - } - public int incrementRefCount(final PagingStore pagingStore, final MessageReference reference) throws Exception { int count = refCount.incrementAndGet(); Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -324,7 +324,7 @@ remotingConnection.removeFailureListener(this); // Return any outstanding credits - + closed = true; for (CreditManagerHolder holder : creditManagerHolders.values()) @@ -648,7 +648,7 @@ { log.error("Failed to query consumer deliveries", e); } - + sendResponse(message, null, false, false); } @@ -1064,9 +1064,9 @@ response = new SessionXAResponseMessage(true, XAException.XAER_PROTO, msg); } else - { + { Transaction theTx = resourceManager.removeTransaction(xid); - + if (theTx == null) { // checked heuristic committed transactions @@ -1436,9 +1436,9 @@ Packet response = null; ServerMessage message = packet.getServerMessage(); - + try - { + { long id = storageManager.generateUniqueID(); message.setMessageID(id); @@ -1510,9 +1510,9 @@ currentLargeMessage = null; message.releaseResources(); - + send(message); - + releaseOutStanding(message); } @@ -1857,7 +1857,7 @@ private void doRollback(final boolean lastMessageAsDelived, final Transaction theTx) throws Exception { boolean wasStarted = started; - + List<MessageReference> toCancel = new ArrayList<MessageReference>(); for (ServerConsumer consumer : consumers.values()) @@ -1874,7 +1874,7 @@ { ref.getQueue().cancel(theTx, ref); } - + theTx.rollback(); if (wasStarted) @@ -1899,7 +1899,7 @@ tx = new TransactionImpl(storageManager); } - + /* * The way flow producer flow control works is as follows: * The client can only send messages as long as it has credits. It requests credits from the server @@ -1914,11 +1914,11 @@ private void releaseOutStanding(final ServerMessage message) throws Exception { CreditManagerHolder holder = getCreditManagerHolder(message.getDestination()); - + int size = message.getEncodeSize(); - + holder.outstandingCredits -= size; - + holder.store.returnProducerCredits(size); } @@ -1945,6 +1945,6 @@ else { postOffice.route(msg, tx); - } + } } } Modified: trunk/src/main/org/hornetq/core/transaction/impl/TransactionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/transaction/impl/TransactionImpl.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/src/main/org/hornetq/core/transaction/impl/TransactionImpl.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -62,7 +62,7 @@ private final long createTime; public TransactionImpl(final StorageManager storageManager) - { + { this.storageManager = storageManager; xid = null; @@ -73,7 +73,7 @@ } public TransactionImpl(final Xid xid, final StorageManager storageManager, final PostOffice postOffice) - { + { this.storageManager = storageManager; this.xid = xid; @@ -84,7 +84,7 @@ } public TransactionImpl(final long id, final Xid xid, final StorageManager storageManager) - { + { this.storageManager = storageManager; this.xid = xid; @@ -276,7 +276,7 @@ if (operations != null) { for (TransactionOperation operation : operations) - { + { operation.beforeRollback(this); } } Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -37,6 +37,7 @@ import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.tests.integration.largemessage.LargeMessageTestBase; +import org.hornetq.tests.util.RandomUtil; import org.hornetq.utils.DataConstants; import org.hornetq.utils.SimpleString; @@ -436,8 +437,10 @@ consumerExpiry = session.createConsumer(ADDRESS_DLA); msg1 = consumerExpiry.receive(5000); + assertNotNull(msg1); - // msg1.acknowledge(); + + msg1.acknowledge(); for (int i = 0; i < messageSize; i++) { @@ -1348,19 +1351,28 @@ } - public void testSendRollbackXA() throws Exception + public void testSendRollbackXADurable() throws Exception { - internalTestSendRollback(true); + internalTestSendRollback(true, true); } + + public void testSendRollbackXANonDurable() throws Exception + { + internalTestSendRollback(true, false); + } - public void testSendRollback() throws Exception + public void testSendRollbackDurable() throws Exception { - internalTestSendRollback(false); + internalTestSendRollback(false, true); } + + public void testSendRollbackNonDurable() throws Exception + { + internalTestSendRollback(false, false); + } - private void internalTestSendRollback(final boolean isXA) throws Exception + private void internalTestSendRollback(final boolean isXA, final boolean durable) throws Exception { - ClientSession session = null; try @@ -1379,13 +1391,13 @@ if (isXA) { - xid = newXID(); + xid = RandomUtil.randomXid(); session.start(xid, XAResource.TMNOFLAGS); } ClientProducer producer = session.createProducer(ADDRESS); - Message clientFile = createLargeClientMessage(session, 50000, false); + Message clientFile = createLargeClientMessage(session, 50000, durable); for (int i = 0; i < 1; i++) { Deleted: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCleanupTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCleanupTest.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCleanupTest.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -1,196 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.tests.integration.largemessage; - -import java.io.File; -import java.io.FileOutputStream; -import java.util.HashMap; -import java.util.concurrent.atomic.AtomicInteger; - -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.config.TransportConfiguration; -import org.hornetq.core.exception.HornetQException; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.remoting.impl.RemotingConnectionImpl; -import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; -import org.hornetq.core.remoting.spi.HornetQBuffer; -import org.hornetq.core.settings.impl.AddressSettings; -import org.hornetq.tests.integration.largemessage.mock.MockConnector; -import org.hornetq.tests.integration.largemessage.mock.MockConnectorFactory; - -/** - * A LargeMessageCleanupTest - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class LargeMessageCleanupTest extends LargeMessageTestBase -{ - // Constants ----------------------------------------------------- - - private static final Logger log = Logger.getLogger(LargeMessageCleanupTest.class); - - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - - public void testCleanup() throws Exception - { - clearData(); - - FileOutputStream fileOut = new FileOutputStream(new File(getLargeMessagesDir(), "1234.tmp")); - - fileOut.write(new byte[1024]); // anything - - fileOut.close(); - - Configuration config = createDefaultConfig(); - - server = createServer(true, config, -1, -1, new HashMap<String, AddressSettings>()); - - server.start(); - - try - { - - File directoryLarge = new File(getLargeMessagesDir()); - - assertEquals("The startup should have been deleted 1234.tmp", 0, directoryLarge.list().length); - } - finally - { - server.stop(); - } - } - - public void testFailureOnSendingFile() throws Exception - { - clearData(); - - Configuration config = createDefaultConfig(); - - server = createServer(true, config, 10 * 1024, 20 * 1024, new HashMap<String, AddressSettings>()); - - server.start(); - - final int numberOfBytes = 2 * 1024 * 1024; - - ClientSession session = null; - - class LocalCallback implements MockConnector.MockCallback - { - AtomicInteger counter = new AtomicInteger(0); - - ClientSession session; - - public void onWrite(final HornetQBuffer buffer) - { - log.info("calling cb onwrite** "); - if (counter.incrementAndGet() == 5) - { - RemotingConnectionImpl conn = (RemotingConnectionImpl)((ClientSessionInternal)session).getConnection(); - RemotingServiceImpl remotingServiceImpl = (RemotingServiceImpl)server.getRemotingService(); - remotingServiceImpl.connectionException(conn.getID(), - new HornetQException(HornetQException.NOT_CONNECTED, "blah!")); - conn.fail(new HornetQException(HornetQException.NOT_CONNECTED, "blah")); - throw new IllegalStateException("blah"); - } - } - } - - LocalCallback callback = new LocalCallback(); - - try - { - HashMap<String, Object> parameters = new HashMap<String, Object>(); - parameters.put("callback", callback); - - TransportConfiguration transport = new TransportConfiguration(MockConnectorFactory.class.getCanonicalName(), - parameters); - - ClientSessionFactory mockFactory = new ClientSessionFactoryImpl(transport); - - mockFactory.setBlockOnNonPersistentSend(false); - mockFactory.setBlockOnPersistentSend(false); - mockFactory.setBlockOnAcknowledge(false); - - session = mockFactory.createSession(null, null, false, true, true, false, 0); - - callback.session = session; - - session.createQueue(ADDRESS, ADDRESS, null, true); - - ClientProducer producer = session.createProducer(ADDRESS); - - ClientMessage clientLarge = createLargeClientMessage(session, numberOfBytes); - - try - { - producer.send(clientLarge); - - fail("Exception was expected!"); - } - catch (Exception e) - { - } - - validateNoFilesOnLargeDir(); - - session.close(); - } - finally - { - try - { - server.stop(); - } - catch (Exception ignored) - { - ignored.printStackTrace(); - } - } - - } - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - protected void setUp() throws Exception - { - super.setUp(); - } - - protected void tearDown() throws Exception - { - super.tearDown(); - } - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Deleted: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -1,567 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.tests.integration.largemessage; - -import java.lang.management.ManagementFactory; -import java.util.concurrent.Executor; - -import javax.transaction.xa.XAResource; -import javax.transaction.xa.Xid; - -import org.hornetq.core.buffers.ChannelBuffers; -import org.hornetq.core.client.ClientConsumer; -import org.hornetq.core.client.ClientMessage; -import org.hornetq.core.client.ClientProducer; -import org.hornetq.core.client.ClientSession; -import org.hornetq.core.client.ClientSessionFactory; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.config.Configuration; -import org.hornetq.core.persistence.StorageManager; -import org.hornetq.core.persistence.impl.journal.JournalLargeServerMessage; -import org.hornetq.core.persistence.impl.journal.JournalStorageManager; -import org.hornetq.core.security.HornetQSecurityManager; -import org.hornetq.core.security.impl.HornetQSecurityManagerImpl; -import org.hornetq.core.server.HornetQServer; -import org.hornetq.core.server.LargeServerMessage; -import org.hornetq.core.server.impl.HornetQServerImpl; -import org.hornetq.core.settings.impl.AddressSettings; -import org.hornetq.tests.util.ServiceTestBase; -import org.hornetq.tests.util.SpawnedVMSupport; - -/** - * A LargeMessageCrashTest - * - * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class LargeMessageCrashTest extends ServiceTestBase -{ - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - static String QUEUE_NAME = "MY-QUEUE"; - - static int LARGE_MESSAGE_SIZE = 5 * ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE; - - static int PAGED_MESSAGE_SIZE = 1024; - - static int NUMBER_OF_PAGES_MESSAGES = 100; - - boolean failAfterRename; - - // Static -------------------------------------------------------- - - public static void main(String args[]) - { - LargeMessageCrashTest serverTest = new LargeMessageCrashTest(); - - serverTest.failAfterRename = false; - - for (String arg : args) - { - if (arg.equals("failAfterRename")) - { - serverTest.failAfterRename = true; - } - } - - for (String arg : args) - { - if (arg.equals("remoteJournalSendNonTransactional")) - { - serverTest.remoteJournalSendNonTransactional(); - } - else if (arg.equals("remoteJournalSendTransactional")) - { - serverTest.remoteJournalSendTransactional(); - } - else if (arg.equals("remotePreparedTransaction")) - { - serverTest.remotePreparedTransaction(); - } - else if (arg.equals("remotePaging")) - { - serverTest.remotePaging(); - } - } - } - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Commented out for https://jira.jboss.org/jira/browse/HORNETQ-49 - public void testFoo() - { - - } - -// public void testJournalSendNonTransactional1() throws Exception -// { -// internalTestSend(false, false); -// } -// -// public void testJournalSendNonTransactional2() throws Exception -// { -// internalTestSend(true, false); -// } -// -// public void testJournalSendTransactional1() throws Exception -// { -// internalTestSend(false, true); -// } -// -// public void testJournalSendTransactional2() throws Exception -// { -// internalTestSend(true, true); -// } - - public void internalTestSend(boolean failureAfterRename, boolean transactional) throws Exception - { - if (transactional) - { - runExternalProcess(failureAfterRename, "remoteJournalSendTransactional"); - } - else - { - runExternalProcess(failureAfterRename, "remoteJournalSendNonTransactional"); - } - - HornetQServer server = newServer(false); - - try - { - server.start(); - - ClientSessionFactory cf = createInVMFactory(); - - ClientSession session = cf.createSession(true, true); - - ClientConsumer cons = session.createConsumer(QUEUE_NAME); - - session.start(); - - assertNull(cons.receive(100)); - - session.close(); - - validateNoFilesOnLargeDir(); - } - finally - { - try - { - server.stop(); - } - catch (Throwable ignored) - { - } - } - } - - public void testPreparedTransaction() throws Exception - { - runExternalProcess(false, "remotePreparedTransaction"); - - HornetQServer server = newServer(false); - - server.start(); - - ClientSessionFactory cf = createInVMFactory(); - - ClientSession session = cf.createSession(true, false, false); - - Xid[] xids = session.recover(XAResource.TMSTARTRSCAN); - - assertEquals(1, xids.length); - - session.rollback(xids[0]); - - session.close(); - - server.stop(); - - validateNoFilesOnLargeDir(); - - } - - public void testPreparedTransactionAndCommit() throws Exception - { - runExternalProcess(false, "remotePreparedTransaction"); - - HornetQServer server = newServer(false); - - server.start(); - - ClientSessionFactory cf = createInVMFactory(); - - ClientSession session = cf.createSession(true, false, false); - - Xid[] xids = session.recover(XAResource.TMSTARTRSCAN); - - assertEquals(1, xids.length); - - session.commit(xids[0], false); - - session.close(); - - session = cf.createSession(false, false); - - ClientConsumer consumer = session.createConsumer(QUEUE_NAME); - - session.start(); - - ClientMessage msg = consumer.receive(5000); - - assertNotNull(msg); - - msg.acknowledge(); - - for (int i = 0; i < LARGE_MESSAGE_SIZE; i++) - { - assertEquals(getSamplebyte(i), msg.getBody().readByte()); - } - - session.commit(); - - session.close(); - - server.stop(); - - validateNoFilesOnLargeDir(); - - } - - - public void testPaging() throws Exception - { - runExternalProcess(false, "remotePaging"); - - HornetQServer server = newServer(false); - - server.start(); - - ClientSessionFactory cf = createInVMFactory(); - - ClientSession session = cf.createSession(false, true, true); - - ClientConsumer consumer = session.createConsumer(QUEUE_NAME); - - session.start(); - - for (int i = 0; i < NUMBER_OF_PAGES_MESSAGES; i++) - { - ClientMessage msg = consumer.receive(50000); - assertNotNull(msg); - msg.acknowledge(); - session.commit(); - } - - ClientMessage msg = consumer.receiveImmediate(); - assertNull(msg); - - session.close(); - - server.stop(); - - validateNoFilesOnLargeDir(); - - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - /** - * @throws Exception - * @throws InterruptedException - */ - private void runExternalProcess(boolean failAfterRename, String methodName) throws Exception, InterruptedException - { - System.err.println("running external process..."); - - Process process = SpawnedVMSupport.spawnVM(this.getClass().getCanonicalName(), - "-Xms128m -Xmx128m ", - new String[] {}, - true, - true, - methodName, - (failAfterRename ? "failAfterRename" : "regularFail")); - - assertEquals(100, process.waitFor()); - } - - // Inner classes ------------------------------------------------- - - public void remoteJournalSendNonTransactional() - { - - try - { - startServer(failAfterRename, true); - - ClientSessionFactory factory = createInVMFactory(); - ClientSession session = factory.createSession(true, true); - - try - { - session.createQueue(QUEUE_NAME, QUEUE_NAME, true); - } - catch (Throwable ignored) - { - } - - ClientProducer prod = session.createProducer(QUEUE_NAME); - - prod.send(createLargeClientMessage(session, LARGE_MESSAGE_SIZE, true)); - } - catch (Exception e) - { - e.printStackTrace(); - System.exit(-1); - } - - } - - public void remoteJournalSendTransactional() - { - try - { - startServer(failAfterRename, true); - - ClientSessionFactory factory = createInVMFactory(); - ClientSession session = factory.createSession(false, false); - - try - { - session.createQueue(QUEUE_NAME, QUEUE_NAME, true); - } - catch (Throwable ignored) - { - } - - ClientProducer prod = session.createProducer(QUEUE_NAME); - - prod.send(createLargeClientMessage(session, LARGE_MESSAGE_SIZE, true)); - } - catch (Exception e) - { - e.printStackTrace(); - System.exit(-1); - } - - } - - public void remotePreparedTransaction() - { - try - { - startServer(failAfterRename, false); - - ClientSessionFactory factory = createInVMFactory(); - ClientSession session = factory.createSession(true, false, false); - - try - { - session.createQueue(QUEUE_NAME, QUEUE_NAME, true); - } - catch (Throwable ignored) - { - } - - ClientProducer prod = session.createProducer(QUEUE_NAME); - - Xid xid = newXID(); - session.start(xid, XAResource.TMNOFLAGS); - - prod.send(createLargeClientMessage(session, LARGE_MESSAGE_SIZE, true)); - - session.end(xid, XAResource.TMSUCCESS); - session.prepare(xid); - - Runtime.getRuntime().halt(100); - } - catch (Exception e) - { - e.printStackTrace(); - System.exit(-1); - } - - } - - public void remotePaging() - { - try - { - startServer(failAfterRename, true); - - ClientSessionFactory factory = createInVMFactory(); - ClientSession session = factory.createSession(false, false, false); - - try - { - session.createQueue(QUEUE_NAME, QUEUE_NAME, true); - } - catch (Throwable ignored) - { - } - - ClientProducer prod = session.createProducer(QUEUE_NAME); - - byte body[] = new byte[PAGED_MESSAGE_SIZE]; - for (int i = 0; i < body.length; i++) - { - body[i] = getSamplebyte(i); - } - - ClientMessage msg = session.createClientMessage(true); - - msg.setBody(ChannelBuffers.wrappedBuffer(body)); - - for (int i = 0; i < NUMBER_OF_PAGES_MESSAGES; i++) - { - prod.send(msg); - } - - session.commit(); - - session.close(); - - session = factory.createSession(false, true, true); - prod = session.createProducer(QUEUE_NAME); - - prod.send(createLargeClientMessage(session, LARGE_MESSAGE_SIZE, true)); - - Runtime.getRuntime().halt(100); - } - catch (Exception e) - { - e.printStackTrace(); - System.exit(-1); - } - - } - - protected ClientMessage createLargeClientMessage(final ClientSession session, - final long numberOfBytes, - final boolean persistent) throws Exception - { - - ClientMessage clientMessage = session.createClientMessage(persistent); - - clientMessage.setBodyInputStream(createFakeLargeStream(numberOfBytes)); - - return clientMessage; - } - - protected void startServer(boolean failAfterRename, boolean fail) - { - this.failAfterRename = failAfterRename; - try - { - HornetQServer server = newServer(fail); - server.start(); - } - catch (Exception e) - { - e.printStackTrace(); - } - } - - private HornetQServer newServer(boolean failing) - { - Configuration configuration = createDefaultConfig(false); - HornetQSecurityManager securityManager = new HornetQSecurityManagerImpl(); - - HornetQServer server; - - if (failing) - { - server = new FailingHornetQServer(configuration, securityManager); - } - else - { - server = new HornetQServerImpl(configuration, securityManager); - } - - AddressSettings defaultSetting = new AddressSettings(); - defaultSetting.setPageSizeBytes(10 * 1024); - defaultSetting.setMaxSizeBytes(100 * 1024); - - server.getAddressSettingsRepository().addMatch("#", defaultSetting); - - return server; - } - - /** This is hacking HornetQServerImpl, - * to make sure the server will fail right - * before the page-file was removed */ - private class FailingHornetQServer extends HornetQServerImpl - { - FailingHornetQServer(final Configuration config, final HornetQSecurityManager securityManager) - { - super(config, ManagementFactory.getPlatformMBeanServer(), securityManager); - } - - @Override - protected StorageManager createStorageManager() - { - return new FailingStorageManager(getConfiguration(), getExecutor()); - } - - } - - private class FailingStorageManager extends JournalStorageManager - { - - public FailingStorageManager(final Configuration config, final Executor executor) - { - super(config, executor); - } - - @Override - public LargeServerMessage createLargeMessage() - { - return new FailinJournalLargeServerMessage(this); - } - - } - - private class FailinJournalLargeServerMessage extends JournalLargeServerMessage - { - /** - * @param storageManager - */ - public FailinJournalLargeServerMessage(final JournalStorageManager storageManager) - { - super(storageManager); - } - - @Override - public void setStored() throws Exception - { - if (failAfterRename) - { - super.setStored(); - } - Runtime.getRuntime().halt(100); - } - - } - -} Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-11-02 23:23:19 UTC (rev 8188) +++ trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-11-03 08:45:42 UTC (rev 8189) @@ -248,7 +248,7 @@ manager.largeMessageWrite(500, new byte[1024]); - manager.largeMessageEnd(500); + manager.largeMessageDelete(500); blockOnReplication(manager);

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8188 - trunk/tests/src/org/hornetq/tests/integration/largemessage.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-02 18:23:19 -0500 (Mon, 02 Nov 2009) New Revision: 8188 Modified: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java Log: commented out test since it's bogus Modified: trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java 2009-11-02 19:18:41 UTC (rev 8187) +++ trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageCrashTest.java 2009-11-02 23:23:19 UTC (rev 8188) @@ -104,26 +104,32 @@ // Public -------------------------------------------------------- - public void testJournalSendNonTransactional1() throws Exception + // Commented out for https://jira.jboss.org/jira/browse/HORNETQ-49 + public void testFoo() { - internalTestSend(false, false); + } + +// public void testJournalSendNonTransactional1() throws Exception +// { +// internalTestSend(false, false); +// } +// +// public void testJournalSendNonTransactional2() throws Exception +// { +// internalTestSend(true, false); +// } +// +// public void testJournalSendTransactional1() throws Exception +// { +// internalTestSend(false, true); +// } +// +// public void testJournalSendTransactional2() throws Exception +// { +// internalTestSend(true, true); +// } - public void testJournalSendNonTransactional2() throws Exception - { - internalTestSend(true, false); - } - - public void testJournalSendTransactional1() throws Exception - { - internalTestSend(false, true); - } - - public void testJournalSendTransactional2() throws Exception - { - internalTestSend(true, true); - } - public void internalTestSend(boolean failureAfterRename, boolean transactional) throws Exception { if (transactional)

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8187 - trunk/src/main/org/hornetq/core/client/impl.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-02 14:18:41 -0500 (Mon, 02 Nov 2009) New Revision: 8187 Modified: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java Log: tweak Modified: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2009-11-02 17:46:47 UTC (rev 8186) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2009-11-02 19:18:41 UTC (rev 8187) @@ -257,7 +257,7 @@ { Version clientVersion = VersionLoader.getVersion(); - RemotingConnection connection = null; + RemotingConnection theConnection = null; Lock lock = null; @@ -267,9 +267,9 @@ synchronized (failoverLock) { - connection = getConnectionWithRetry(reconnectAttempts); + theConnection = getConnectionWithRetry(reconnectAttempts); - if (connection == null) + if (theConnection == null) { if (exitLoop) { @@ -282,7 +282,7 @@ } - channel1 = connection.getChannel(1, -1); + channel1 = theConnection.getChannel(1, -1); // Lock it - this must be done while the failoverLock is held channel1.getLock().lock(); @@ -296,7 +296,7 @@ inCreateSession = true; } - long sessionChannelID = connection.generateChannelID(); + long sessionChannelID = theConnection.generateChannelID(); Packet request = new CreateSessionMessage(name, sessionChannelID, @@ -322,10 +322,6 @@ // This means the thread was blocked on create session and failover unblocked it // so failover could occur - // So we just need to return our connections and flag for retry - - checkCloseConnection(); - retry = true; continue; @@ -337,8 +333,8 @@ } CreateSessionResponseMessage response = (CreateSessionResponseMessage)pResponse; - - Channel sessionChannel = connection.getChannel(sessionChannelID, + + Channel sessionChannel = theConnection.getChannel(sessionChannelID, confWindowSize); ClientSessionInternal session = new ClientSessionImpl(this, @@ -361,7 +357,7 @@ blockOnPersistentSend, cacheLargeMessageClient, minLargeMessageSize, - connection, + theConnection, response.getServerVersion(), sessionChannel, orderedExecutorFactory.getExecutor()); @@ -383,8 +379,6 @@ lock = null; } - checkCloseConnection(); - if (t instanceof HornetQException) { throw (HornetQException)t; @@ -613,11 +607,11 @@ backupTransportParams = null; - reconnectSessions(reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); + reconnectSessions(oldConnection, reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); } else { - reconnectSessions(reconnectAttempts); + reconnectSessions(oldConnection, reconnectAttempts); } oldConnection.destroy(); @@ -663,8 +657,8 @@ /* * Re-attach sessions all pre-existing sessions to the new remoting connection */ - private void reconnectSessions(final int reconnectAttempts) - { + private void reconnectSessions(final RemotingConnection oldConnection, final int reconnectAttempts) + { RemotingConnection backupConnection = getConnectionWithRetry(reconnectAttempts); if (backupConnection == null) @@ -673,9 +667,9 @@ return; } - - List<FailureListener> oldListeners = connection.getFailureListeners(); - + + List<FailureListener> oldListeners = oldConnection.getFailureListeners(); + List<FailureListener> newListeners = new ArrayList<FailureListener>(backupConnection.getFailureListeners()); for (FailureListener listener : oldListeners) @@ -709,9 +703,9 @@ return null; } - RemotingConnection connection = getConnection(); + RemotingConnection theConnection = getConnection(); - if (connection == null) + if (theConnection == null) { // Failed to get connection @@ -751,7 +745,7 @@ } else { - return connection; + return theConnection; } } } @@ -764,7 +758,7 @@ { pingRunnable.cancel(); - boolean ok = pingerFuture.cancel(false); + pingerFuture.cancel(false); pingRunnable = null;

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8186 - in trunk: src/main/org/hornetq/core/management/impl and 2 other directories.

by do-not-reply＠jboss.org

Author: timfox Date: 2009-11-02 12:46:47 -0500 (Mon, 02 Nov 2009) New Revision: 8186 Modified: trunk/src/main/org/hornetq/core/exception/HornetQException.java trunk/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java Log: various tweaks including fixing bridge reconnect test Modified: trunk/src/main/org/hornetq/core/exception/HornetQException.java =================================================================== --- trunk/src/main/org/hornetq/core/exception/HornetQException.java 2009-11-02 16:10:49 UTC (rev 8185) +++ trunk/src/main/org/hornetq/core/exception/HornetQException.java 2009-11-02 17:46:47 UTC (rev 8186) @@ -61,6 +61,8 @@ public static final int LARGE_MESSAGE_ERROR_BODY = 110; public static final int TRANSACTION_ROLLED_BACK = 111; + + public static final int SESSION_CREATION_REJECTED = 112; // Native Error codes ---------------------------------------------- Modified: trunk/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java 2009-11-02 16:10:49 UTC (rev 8185) +++ trunk/src/main/org/hornetq/core/management/impl/ManagementServiceImpl.java 2009-11-02 17:46:47 UTC (rev 8186) @@ -695,7 +695,7 @@ } // start sending notification *messages* only when the *remoting service* if started - if (messagingServer == null || !messagingServer.isStarted() || + if (messagingServer == null || !messagingServer.getRemotingService().isStarted()) { return; Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-11-02 16:10:49 UTC (rev 8185) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-11-02 17:46:47 UTC (rev 8186) @@ -49,10 +49,8 @@ private final Channel channel1; private final RemotingConnection connection; - - public HornetQPacketHandler(final HornetQServer server, - final Channel channel1, - final RemotingConnection connection) + + public HornetQPacketHandler(final HornetQServer server, final Channel channel1, final RemotingConnection connection) { this.server = server; @@ -64,11 +62,11 @@ public void handlePacket(final Packet packet) { byte type = packet.getType(); - + // All these operations need to be idempotent since they are outside of the session // reliability replay functionality switch (type) - { + { case CREATESESSION: { CreateSessionMessage request = (CreateSessionMessage)packet; @@ -76,7 +74,7 @@ handleCreateSession(request); break; - } + } case REATTACH_SESSION: { ReattachSessionMessage request = (ReattachSessionMessage)packet; @@ -90,21 +88,21 @@ // Create queue can also be fielded here in the case of a replicated store and forward queue creation CreateQueueMessage request = (CreateQueueMessage)packet; - + handleCreateQueue(request); break; - } + } case CREATE_REPLICATION: { // Create queue can also be fielded here in the case of a replicated store and forward queue creation CreateReplicationSessionMessage request = (CreateReplicationSessionMessage)packet; - + handleCreateReplication(request); break; - } + } default: { log.error("Invalid packet " + packet); @@ -118,7 +116,7 @@ try { response = server.createSession(request.getName(), - request.getSessionChannelID(), + request.getSessionChannelID(), request.getUsername(), request.getPassword(), request.getMinLargeMessageSize(), @@ -129,11 +127,6 @@ request.isPreAcknowledge(), request.isXA(), request.getWindowSize()); - - if (response == null) - { - response = new HornetQExceptionMessage(new HornetQException(HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS)); - } } catch (Exception e) { @@ -148,10 +141,10 @@ response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR)); } } - - channel1.send(response); + + channel1.send(response); } - + private void handleReattachSession(final ReattachSessionMessage request) { Packet response; @@ -181,14 +174,18 @@ { try { - server.createQueue(request.getAddress(), request.getQueueName(), request.getFilterString(), request.isDurable(), request.isTemporary()); + server.createQueue(request.getAddress(), + request.getQueueName(), + request.getFilterString(), + request.isDurable(), + request.isTemporary()); } catch (Exception e) { log.error("Failed to handle create queue", e); } } - + private void handleCreateReplication(final CreateReplicationSessionMessage request) { Packet response; @@ -196,17 +193,17 @@ try { Channel channel = connection.getChannel(request.getSessionChannelID(), request.getWindowSize()); - + ReplicationEndpoint endpoint = server.createReplicationEndpoint(channel); - + channel.setHandler(endpoint); - + response = new NullResponseMessage(); } - catch (Exception e) + catch (Exception e) { log.warn(e.getMessage(), e); - + if (e instanceof HornetQException) { response = new HornetQExceptionMessage((HornetQException)e); @@ -219,9 +216,5 @@ channel1.send(response); } - - - - } \ No newline at end of file Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-02 16:10:49 UTC (rev 8185) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-02 17:46:47 UTC (rev 8186) @@ -135,9 +135,9 @@ // Attributes // ----------------------------------------------------------------------------------- - private SimpleString nodeID; + private volatile SimpleString nodeID; - private UUID uuid; + private volatile UUID uuid; private final Version version; @@ -149,39 +149,39 @@ private volatile boolean started; - private SecurityStore securityStore; + private volatile SecurityStore securityStore; private final HierarchicalRepository<AddressSettings> addressSettingsRepository; - private QueueFactory queueFactory; + private volatile QueueFactory queueFactory; - private PagingManager pagingManager; + private volatile PagingManager pagingManager; - private PostOffice postOffice; + private volatile PostOffice postOffice; - private ExecutorService threadPool; + private volatile ExecutorService threadPool; - private ScheduledExecutorService scheduledPool; + private volatile ScheduledExecutorService scheduledPool; - private ExecutorFactory executorFactory; + private volatile ExecutorFactory executorFactory; - private HierarchicalRepository<Set<Role>> securityRepository; + private volatile HierarchicalRepository<Set<Role>> securityRepository; - private ResourceManager resourceManager; + private volatile ResourceManager resourceManager; - private HornetQServerControlImpl messagingServerControl; + private volatile HornetQServerControlImpl messagingServerControl; - private ClusterManager clusterManager; + private volatile ClusterManager clusterManager; - private StorageManager storageManager; + private volatile StorageManager storageManager; - private RemotingService remotingService; + private volatile RemotingService remotingService; - private ManagementService managementService; + private volatile ManagementService managementService; private MemoryManager memoryManager; - private DeploymentManager deploymentManager; + private volatile DeploymentManager deploymentManager; private Deployer basicUserCredentialsDeployer; @@ -205,7 +205,7 @@ private final Set<ActivateCallback> activateCallbacks = new HashSet<ActivateCallback>(); - private GroupingHandler groupingHandler; + private volatile GroupingHandler groupingHandler; // Constructors // --------------------------------------------------------------------------------- @@ -258,8 +258,6 @@ this.addressSettingsRepository = new HierarchicalObjectRepository<AddressSettings>(); addressSettingsRepository.setDefault(new AddressSettings()); - - // this.managementConnectorID = managementConnectorSequence.decrementAndGet(); } // lifecycle methods @@ -317,92 +315,130 @@ super.finalize(); } - public synchronized void stop() throws Exception + public void stop() throws Exception { - if (!started) + synchronized (this) { - return; - } + if (!started) + { + return; + } - if (clusterManager != null) - { - clusterManager.stop(); - } + if (clusterManager != null) + { + clusterManager.stop(); + } - if (groupingHandler != null) - { - managementService.removeNotificationListener(groupingHandler); - groupingHandler = null; - } - // Need to flush all sessions to make sure all confirmations get sent back to client + if (groupingHandler != null) + { + managementService.removeNotificationListener(groupingHandler); + groupingHandler = null; + } + // Need to flush all sessions to make sure all confirmations get sent back to client - for (ServerSession session : sessions.values()) - { - session.getChannel().flushConfirmations(); - } + for (ServerSession session : sessions.values()) + { + session.getChannel().flushConfirmations(); + } - remotingService.stop(); + remotingService.stop(); - // Stop the deployers - if (configuration.isFileDeploymentEnabled()) - { - basicUserCredentialsDeployer.stop(); + // Stop the deployers + if (configuration.isFileDeploymentEnabled()) + { + basicUserCredentialsDeployer.stop(); - addressSettingsDeployer.stop(); + addressSettingsDeployer.stop(); - if (queueDeployer != null) + if (queueDeployer != null) + { + queueDeployer.stop(); + } + + if (securityDeployer != null) + { + securityDeployer.stop(); + } + + deploymentManager.stop(); + } + + managementService.unregisterServer(); + + managementService.stop(); + + if (storageManager != null) { - queueDeployer.stop(); + storageManager.stop(); } - if (securityDeployer != null) + if (replicationEndpoint != null) { - securityDeployer.stop(); + replicationEndpoint.stop(); + replicationEndpoint = null; } - deploymentManager.stop(); - } + if (securityManager != null) + { + securityManager.stop(); + } - managementService.unregisterServer(); + if (resourceManager != null) + { + resourceManager.stop(); + } - managementService.stop(); + if (postOffice != null) + { + postOffice.stop(); + } - if (storageManager != null) - { - storageManager.stop(); - } + // Need to shutdown pools before shutting down paging manager to make sure everything is written ok - if (replicationEndpoint != null) - { - replicationEndpoint.stop(); - replicationEndpoint = null; - } + List<Runnable> tasks = scheduledPool.shutdownNow(); - if (securityManager != null) - { - securityManager.stop(); - } + for (Runnable task : tasks) + { + log.debug("Waiting for " + task); + } - if (resourceManager != null) - { - resourceManager.stop(); - } + threadPool.shutdown(); - if (postOffice != null) - { - postOffice.stop(); - } + scheduledPool = null; - // Need to shutdown pools before shutting down paging manager to make sure everything is written ok + if (pagingManager != null) + { + pagingManager.stop(); + } - List<Runnable> tasks = scheduledPool.shutdownNow(); + if (memoryManager != null) + { + memoryManager.stop(); + } - for (Runnable task : tasks) - { - log.debug("Waiting for " + task); + pagingManager = null; + securityStore = null; + resourceManager = null; + postOffice = null; + securityRepository = null; + securityStore = null; + queueFactory = null; + resourceManager = null; + messagingServerControl = null; + memoryManager = null; + + sessions.clear(); + + started = false; + initialised = false; + uuid = null; + nodeID = null; + + log.info("HornetQ Server version " + getVersion().getFullVersion() + " stopped"); + + Logger.reset(); } - threadPool.shutdown(); try { if (!threadPool.awaitTermination(30000, TimeUnit.MILLISECONDS)) @@ -414,41 +450,7 @@ { // Ignore } - - scheduledPool = null; threadPool = null; - - if (pagingManager != null) - { - pagingManager.stop(); - } - - if (memoryManager != null) - { - memoryManager.stop(); - } - - pagingManager = null; - securityStore = null; - resourceManager = null; - postOffice = null; - securityRepository = null; - securityStore = null; - queueFactory = null; - resourceManager = null; - messagingServerControl = null; - memoryManager = null; - - sessions.clear(); - - started = false; - initialised = false; - uuid = null; - nodeID = null; - - log.info("HornetQ Server version " + getVersion().getFullVersion() + " stopped"); - - Logger.reset(); } // HornetQServer implementation @@ -509,7 +511,7 @@ return version; } - public boolean isStarted() + public synchronized boolean isStarted() { return started; } @@ -520,9 +522,14 @@ } public ReattachSessionResponseMessage reattachSession(final RemotingConnection connection, - final String name, - final int lastReceivedCommandID) throws Exception + final String name, + final int lastReceivedCommandID) throws Exception { + if (!started) + { + return null; + } + ServerSession session = sessions.get(name); if (!checkActivate()) @@ -552,7 +559,7 @@ } sessions.remove(name); - + return new ReattachSessionResponseMessage(-1, false); } else @@ -566,18 +573,23 @@ } public CreateSessionResponseMessage createSession(final String name, - final long channelID, - final String username, - final String password, - final int minLargeMessageSize, - final int incrementingVersion, - final RemotingConnection connection, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final boolean xa, - final int sendWindowSize) throws Exception + final long channelID, + final String username, + final String password, + final int minLargeMessageSize, + final int incrementingVersion, + final RemotingConnection connection, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final boolean xa, + final int sendWindowSize) throws Exception { + if (!started) + { + throw new HornetQException(HornetQException.SESSION_CREATION_REJECTED, "Server not started"); + } + if (version.getIncrementingVersion() != incrementingVersion) { log.warn("Client with version " + incrementingVersion + @@ -588,14 +600,14 @@ ". " + "Please ensure all clients and servers are upgraded to the same version for them to " + "interoperate properly"); - return null; + throw new HornetQException(HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS, "Server and client versions incompatible"); } if (!checkActivate()) { // Backup server is not ready to accept connections - return new CreateSessionResponseMessage(version.getIncrementingVersion()); + throw new HornetQException(HornetQException.SESSION_CREATION_REJECTED, "Server will not accept create session requests"); } if (securityStore != null) @@ -673,7 +685,7 @@ return sessions.get(name); } - public List<ServerSession> getSessions(final String connectionID) + public synchronized List<ServerSession> getSessions(final String connectionID) { Set<Entry<String, ServerSession>> sessionEntries = sessions.entrySet(); List<ServerSession> matchingSessions = new ArrayList<ServerSession>(); @@ -688,11 +700,12 @@ return matchingSessions; } - public Set<ServerSession> getSessions() + public synchronized Set<ServerSession> getSessions() { return new HashSet<ServerSession>(sessions.values()); } + //TODO - should this really be here?? It's only used in tests public boolean isInitialised() { synchronized (initialiseLock) @@ -727,19 +740,19 @@ } public Queue createQueue(final SimpleString address, - final SimpleString queueName, - final SimpleString filterString, - final boolean durable, - final boolean temporary) throws Exception + final SimpleString queueName, + final SimpleString filterString, + final boolean durable, + final boolean temporary) throws Exception { return createQueue(address, queueName, filterString, durable, temporary, false); } public Queue deployQueue(final SimpleString address, - final SimpleString queueName, - final SimpleString filterString, - final boolean durable, - final boolean temporary) throws Exception + final SimpleString queueName, + final SimpleString filterString, + final boolean durable, + final boolean temporary) throws Exception { return createQueue(address, queueName, filterString, durable, temporary, true); } @@ -793,12 +806,12 @@ activateCallbacks.remove(callback); } - public ExecutorFactory getExecutorFactory() + public synchronized ExecutorFactory getExecutorFactory() { return executorFactory; } - public void setGroupingHandler(GroupingHandler groupingHandler) + public void setGroupingHandler(final GroupingHandler groupingHandler) { this.groupingHandler = groupingHandler; } @@ -900,7 +913,7 @@ return true; } - private synchronized void callActivateCallbacks() + private void callActivateCallbacks() { for (ActivateCallback callback : activateCallbacks) { Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java 2009-11-02 16:10:49 UTC (rev 8185) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/MessageRedistributionTest.java 2009-11-02 17:46:47 UTC (rev 8186) @@ -35,16 +35,16 @@ { super.setUp(); - start(); + start(); } - + private void start() throws Exception { setupServers(); - - setRedistributionDelay(0); + + setRedistributionDelay(0); } - + private void stop() throws Exception { stopServers(); @@ -71,7 +71,7 @@ public void testRedistributionWhenConsumerIsClosed() throws Exception { setupCluster(false); - + log.info("Doing test"); startServers(0, 1, 2); @@ -105,7 +105,7 @@ removeConsumer(1); verifyReceiveRoundRobinInSomeOrderWithCounts(false, ids1, 0, 2); - + log.info("Test done"); } @@ -358,100 +358,98 @@ verifyReceiveAll(20, 1); verifyNotReceive(1); } - + public void testBackAndForth() throws Exception { for (int i = 0; i < 10; i++) { setupCluster(false); - + startServers(0, 1, 2); - + setupSessionFactory(0, isNetty()); setupSessionFactory(1, isNetty()); setupSessionFactory(2, isNetty()); - + final String ADDRESS = "queues.testaddress"; final String QUEUE = "queue0"; - - + createQueue(0, ADDRESS, QUEUE, null, false); createQueue(1, ADDRESS, QUEUE, null, false); createQueue(2, ADDRESS, QUEUE, null, false); - + addConsumer(0, 0, QUEUE, null); - + waitForBindings(0, ADDRESS, 1, 1, true); waitForBindings(1, ADDRESS, 1, 0, true); waitForBindings(2, ADDRESS, 1, 0, true); - + waitForBindings(0, ADDRESS, 2, 0, false); waitForBindings(1, ADDRESS, 2, 1, false); waitForBindings(2, ADDRESS, 2, 1, false); - + send(0, ADDRESS, 20, false, null); - + waitForMessages(0, ADDRESS, 20); - + removeConsumer(0); - + waitForBindings(0, ADDRESS, 1, 0, true); waitForBindings(1, ADDRESS, 1, 0, true); waitForBindings(2, ADDRESS, 1, 0, true); - + waitForBindings(0, ADDRESS, 2, 0, false); waitForBindings(1, ADDRESS, 2, 0, false); waitForBindings(2, ADDRESS, 2, 0, false); - + addConsumer(1, 1, QUEUE, null); - + waitForBindings(0, ADDRESS, 1, 0, true); waitForBindings(1, ADDRESS, 1, 1, true); waitForBindings(2, ADDRESS, 1, 0, true); - + waitForMessages(1, ADDRESS, 20); waitForMessages(0, ADDRESS, 0); - - + waitForBindings(0, ADDRESS, 2, 1, false); waitForBindings(1, ADDRESS, 2, 0, false); waitForBindings(2, ADDRESS, 2, 1, false); - + removeConsumer(1); - + waitForBindings(0, ADDRESS, 1, 0, true); waitForBindings(1, ADDRESS, 1, 0, true); waitForBindings(2, ADDRESS, 1, 0, true); - + waitForBindings(0, ADDRESS, 2, 0, false); waitForBindings(1, ADDRESS, 2, 0, false); waitForBindings(2, ADDRESS, 2, 0, false); - + addConsumer(0, 0, QUEUE, null); - + waitForBindings(0, ADDRESS, 1, 1, true); waitForBindings(1, ADDRESS, 1, 0, true); waitForBindings(2, ADDRESS, 1, 0, true); - + waitForBindings(0, ADDRESS, 2, 0, false); waitForBindings(1, ADDRESS, 2, 1, false); waitForBindings(2, ADDRESS, 2, 1, false); - + waitForMessages(0, ADDRESS, 20); - + verifyReceiveAll(20, 0); verifyNotReceive(0); - + addConsumer(1, 1, QUEUE, null); verifyNotReceive(1); removeConsumer(1); - + stop(); start(); } - + } - + public void testRedistributionToQueuesWhereNotAllMessagesMatch() throws Exception { setupCluster(false); @@ -469,7 +467,7 @@ createQueue(1, "queues.testaddress", "queue0", null, false); createQueue(2, "queues.testaddress", "queue0", null, false); - addConsumer(0, 0, "queue0", null); + addConsumer(0, 0, "queue0", null); waitForBindings(0, "queues.testaddress", 1, 1, true); waitForBindings(1, "queues.testaddress", 1, 0, true); @@ -483,18 +481,18 @@ sendInRange(0, "queues.testaddress", 10, 20, false, filter2); removeConsumer(0); - addConsumer(1, 1, "queue0", filter1); - addConsumer(2, 2, "queue0", filter2); + addConsumer(1, 1, "queue0", filter1); + addConsumer(2, 2, "queue0", filter2); verifyReceiveAllInRange(0, 10, 1); verifyReceiveAllInRange(10, 20, 2); } - + public void testDelayedRedistribution() throws Exception { final long delay = 1000; setRedistributionDelay(delay); - + setupCluster(false); startServers(0, 1, 2); @@ -507,7 +505,7 @@ createQueue(1, "queues.testaddress", "queue0", null, false); createQueue(2, "queues.testaddress", "queue0", null, false); - addConsumer(0, 0, "queue0", null); + addConsumer(0, 0, "queue0", null); waitForBindings(0, "queues.testaddress", 1, 1, true); waitForBindings(1, "queues.testaddress", 1, 0, true); @@ -520,20 +518,20 @@ send(0, "queues.testaddress", 20, false, null); long start = System.currentTimeMillis(); - + removeConsumer(0); - addConsumer(1, 1, "queue0", null); - + addConsumer(1, 1, "queue0", null); + long minReceiveTime = start + delay; - + verifyReceiveAllNotBefore(minReceiveTime, 20, 1); } - + public void testDelayedRedistributionCancelled() throws Exception { final long delay = 1000; setRedistributionDelay(delay); - + setupCluster(false); startServers(0, 1, 2); @@ -546,7 +544,7 @@ createQueue(1, "queues.testaddress", "queue0", null, false); createQueue(2, "queues.testaddress", "queue0", null, false); - addConsumer(0, 0, "queue0", null); + addConsumer(0, 0, "queue0", null); waitForBindings(0, "queues.testaddress", 1, 1, true); waitForBindings(1, "queues.testaddress", 1, 0, true); @@ -559,18 +557,18 @@ send(0, "queues.testaddress", 20, false, null); removeConsumer(0); - addConsumer(1, 1, "queue0", null); - + addConsumer(1, 1, "queue0", null); + Thread.sleep(delay / 2); - - //Add it back on the local queue - this should stop any redistributionm - addConsumer(0, 0, "queue0", null); - + + // Add it back on the local queue - this should stop any redistributionm + addConsumer(0, 0, "queue0", null); + Thread.sleep(delay); - + verifyReceiveAll(20, 0); } - + public void testRedistributionNumberOfMessagesGreaterThanBatchSize() throws Exception { setupCluster(false); @@ -585,7 +583,7 @@ createQueue(1, "queues.testaddress", "queue0", null, false); createQueue(2, "queues.testaddress", "queue0", null, false); - addConsumer(0, 0, "queue0", null); + addConsumer(0, 0, "queue0", null); waitForBindings(0, "queues.testaddress", 1, 1, true); waitForBindings(1, "queues.testaddress", 1, 0, true); @@ -598,8 +596,8 @@ send(0, "queues.testaddress", QueueImpl.REDISTRIBUTOR_BATCH_SIZE * 2, false, null); removeConsumer(0); - addConsumer(1, 1, "queue0", null); - + addConsumer(1, 1, "queue0", null); + verifyReceiveAll(QueueImpl.REDISTRIBUTOR_BATCH_SIZE * 2, 1); } @@ -636,7 +634,7 @@ closeAllSessionFactories(); stopServers(0, 1, 2); - + clearServer(0, 1, 2); }

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8185 - trunk/docs/user-manual/en.

by do-not-reply＠jboss.org

Author: jmesnil Date: 2009-11-02 11:10:49 -0500 (Mon, 02 Nov 2009) New Revision: 8185 Modified: trunk/docs/user-manual/en/jms-bridge.xml Log: JMS Bridge documentation * fixed docbook layout Modified: trunk/docs/user-manual/en/jms-bridge.xml =================================================================== --- trunk/docs/user-manual/en/jms-bridge.xml 2009-11-02 16:03:05 UTC (rev 8184) +++ trunk/docs/user-manual/en/jms-bridge.xml 2009-11-02 16:10:49 UTC (rev 8185) @@ -17,7 +17,6 @@   <chapter id="jms-bridge"> - <section> <title>The JMS Bridge</title> <para>HornetQ includes a fully functional JMS message bridge.</para> <para>The function of the bridge is to consume messages from a source queue or topic, and @@ -417,5 +416,4 @@ and use a JMS Bridge between two standalone HornetQ servers.</para> </section> </section> - </section> </chapter>

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8184 - in trunk: examples/javaee/jms-bridge/server and 4 other directories.

by do-not-reply＠jboss.org

Author: jmesnil Date: 2009-11-02 11:03:05 -0500 (Mon, 02 Nov 2009) New Revision: 8184 Added: trunk/src/main/org/hornetq/jms/bridge/JMSBridgeControl.java trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeControlImpl.java Modified: trunk/docs/user-manual/en/jms-bridge.xml trunk/examples/javaee/jms-bridge/server/jms-bridge-jboss-beans.xml trunk/examples/jms/jms-bridge/server1/hornetq-beans.xml trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java trunk/tests/src/org/hornetq/tests/integration/jms/bridge/JMSBridgeTest.java Log: https://jira.jboss.org/jira/browse/HORNETQ-193: register JMS Bridge in JMX to manage it * added JMSBridgeControlMBean to manage JMS Bridge using JMX * added to JMSBridgeImpl ctor optional parameters to inject MBeanServer and MBean's ObjectName * inject JVM platform (resp. AS) MBeanServer in jms/jms-bridge (resp. javaee/jms-bridge) example * doc -- patch contributed by Jose de Castro Modified: trunk/docs/user-manual/en/jms-bridge.xml =================================================================== --- trunk/docs/user-manual/en/jms-bridge.xml 2009-11-02 15:32:07 UTC (rev 8183) +++ trunk/docs/user-manual/en/jms-bridge.xml 2009-11-02 16:03:05 UTC (rev 8184) @@ -98,7 +98,12 @@ <parameter><null /></parameter>  <parameter>true</parameter> - </constructor> +  + <parameter> + <inject bean="MBeanServer"/> + </parameter> + <parameter>org.hornetq:service=JMSBridge</parameter> + </constructor> <property name="transactionManager"> <inject bean="RealTransactionManager"/> </property> @@ -172,6 +177,10 @@ </constructor> </bean> + <bean name="MBeanServer" class="javax.management.MBeanServer"> + <constructor factoryClass="org.jboss.mx.util.MBeanServerLocator" + factoryMethod="locateJBoss"/> + </bean> </deployment></programlisting> <section> <title>JMS Bridge Parameters</title> @@ -318,6 +327,16 @@ it back it will be able to correlate it. </para> </note> </listitem> + <listitem> + <para>MBean Server</para> + <para>To manage the JMS Bridge using JMX, set the MBeanServer where the JMS Bridge MBean + must be registered (e.g. the JVM Platform MBeanServer or JBoss AS MBeanServer)</para> + </listitem> + <listitem> + <para>ObjectName</para> + <para>If you set the MBeanServer, you also need to set the ObjectName used to register + the JMS Bridge MBean (must be unique)</para> + </listitem> </itemizedlist> </section> <section> Modified: trunk/examples/javaee/jms-bridge/server/jms-bridge-jboss-beans.xml =================================================================== --- trunk/examples/javaee/jms-bridge/server/jms-bridge-jboss-beans.xml 2009-11-02 15:32:07 UTC (rev 8183) +++ trunk/examples/javaee/jms-bridge/server/jms-bridge-jboss-beans.xml 2009-11-02 16:03:05 UTC (rev 8184) @@ -48,6 +48,11 @@ <parameter><null /></parameter>  <parameter>true</parameter> +  + <parameter> + <inject bean="MBeanServer"/> + </parameter> + <parameter>org.hornetq:service=JMSBridge</parameter> </constructor> <property name="transactionManager"> <inject bean="RealTransactionManager"/> @@ -118,4 +123,9 @@ </constructor> </bean> + <bean name="MBeanServer" class="javax.management.MBeanServer"> + <constructor factoryClass="org.jboss.mx.util.MBeanServerLocator" + factoryMethod="locateJBoss"/> + </bean> + </deployment> \ No newline at end of file Modified: trunk/examples/jms/jms-bridge/server1/hornetq-beans.xml =================================================================== --- trunk/examples/jms/jms-bridge/server1/hornetq-beans.xml 2009-11-02 15:32:07 UTC (rev 8183) +++ trunk/examples/jms/jms-bridge/server1/hornetq-beans.xml 2009-11-02 16:03:05 UTC (rev 8184) @@ -107,6 +107,11 @@ <parameter><null /></parameter>  <parameter>true</parameter> +  + <parameter> + <inject bean="MBeanServer"/> + </parameter> + <parameter>org.hornetq:service=JMSBridge</parameter> </constructor> <property name="transactionManager"> <inject bean="TransactionManager"/> Added: trunk/src/main/org/hornetq/jms/bridge/JMSBridgeControl.java =================================================================== --- trunk/src/main/org/hornetq/jms/bridge/JMSBridgeControl.java (rev 0) +++ trunk/src/main/org/hornetq/jms/bridge/JMSBridgeControl.java 2009-11-02 16:03:05 UTC (rev 8184) @@ -0,0 +1,94 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.jms.bridge; + +import org.hornetq.core.management.HornetQComponentControl; + +/** + * A JMSBridgeControl + * + * @author <a href="jose(a)voxeo.com">Jose de Castro</a> + * + */ +public interface JMSBridgeControl extends HornetQComponentControl +{ + void pause() throws Exception; + + void resume() throws Exception; + + String getSourceUsername(); + + void setSourceUsername(String name); + + String getSourcePassword(); + + void setSourcePassword(String pwd); + + String getTargetUsername(); + + void setTargetUsername(String name); + + String getTargetPassword(); + + void setTargetPassword(String pwd); + + String getSelector(); + + void setSelector(String selector); + + long getFailureRetryInterval(); + + void setFailureRetryInterval(long interval); + + int getMaxRetries(); + + void setMaxRetries(int retries); + + String getQualityOfServiceMode(); + + void setQualityOfServiceMode(String mode); + + int getMaxBatchSize(); + + void setMaxBatchSize(int size); + + long getMaxBatchTime(); + + void setMaxBatchTime(long time); + + String getSubscriptionName(); + + void setSubscriptionName(String subname); + + String getClientID(); + + void setClientID(String clientID); + + String getTransactionManagerLocatorClass(); + + void setTransactionManagerLocatorClass(String transactionManagerLocatorClass); + + String getTransactionManagerLocatorMethod(); + + void setTransactionManagerLocatorMethod(String transactionManagerLocatorMethod); + + boolean isAddMessageIDInHeader(); + + void setAddMessageIDInHeader(boolean value); + + boolean isPaused(); + + boolean isFailed(); + +} Added: trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeControlImpl.java (rev 0) +++ trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeControlImpl.java 2009-11-02 16:03:05 UTC (rev 8184) @@ -0,0 +1,257 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.jms.bridge.impl; + +import javax.management.StandardMBean; + +import org.hornetq.jms.bridge.JMSBridge; +import org.hornetq.jms.bridge.JMSBridgeControl; +import org.hornetq.jms.bridge.QualityOfServiceMode; + +/** + * A JMSBridgeControlImpl + * + * @author <a href="jose(a)voxeo.com">Jose de Castro</a> + * + */ +public class JMSBridgeControlImpl extends StandardMBean implements JMSBridgeControl +{ + + private JMSBridge bridge; + + // Constructors -------------------------------------------------- + + public JMSBridgeControlImpl(JMSBridge bridge) throws Exception + { + super(JMSBridgeControl.class); + this.bridge = bridge; + } + + // Public -------------------------------------------------------- + + public void pause() throws Exception + { + bridge.pause(); + } + + public void resume() throws Exception + { + bridge.resume(); + } + + public boolean isStarted() + { + return bridge.isStarted(); + } + + public void start() throws Exception + { + bridge.start(); + } + + public void stop() throws Exception + { + bridge.stop(); + } + + public String getClientID() + { + return bridge.getClientID(); + } + + public long getFailureRetryInterval() + { + return bridge.getFailureRetryInterval(); + } + + public int getMaxBatchSize() + { + return bridge.getMaxBatchSize(); + } + + public long getMaxBatchTime() + { + return bridge.getMaxBatchTime(); + } + + public int getMaxRetries() + { + return bridge.getMaxRetries(); + } + + public String getQualityOfServiceMode() + { + QualityOfServiceMode mode = bridge.getQualityOfServiceMode(); + if (mode != null) + { + return mode.name(); + } + else + { + return null; + } + } + + public String getSelector() + { + return bridge.getSelector(); + } + + public String getSourcePassword() + { + return bridge.getSourcePassword(); + } + + public String getSourceUsername() + { + return bridge.getSourceUsername(); + } + + public String getSubscriptionName() + { + return bridge.getSubscriptionName(); + } + + public String getTargetPassword() + { + return bridge.getTargetPassword(); + } + + public String getTargetUsername() + { + return bridge.getTargetUsername(); + } + + public String getTransactionManagerLocatorClass() + { + return bridge.getTransactionManagerLocatorClass(); + } + + public String getTransactionManagerLocatorMethod() + { + return bridge.getTransactionManagerLocatorMethod(); + } + + public boolean isAddMessageIDInHeader() + { + return bridge.isAddMessageIDInHeader(); + } + + public boolean isFailed() + { + return bridge.isFailed(); + } + + public boolean isPaused() + { + return bridge.isPaused(); + } + + public void setAddMessageIDInHeader(boolean value) + { + bridge.setAddMessageIDInHeader(value); + } + + public void setClientID(String clientID) + { + bridge.setClientID(clientID); + } + + public void setFailureRetryInterval(long interval) + { + bridge.setFailureRetryInterval(interval); + } + + public void setMaxBatchSize(int size) + { + bridge.setMaxBatchSize(size); + } + + public void setMaxBatchTime(long time) + { + bridge.setMaxBatchTime(time); + } + + public void setMaxRetries(int retries) + { + bridge.setMaxRetries(retries); + } + + public void setQualityOfServiceMode(String mode) + { + if (mode != null) + { + bridge.setQualityOfServiceMode(QualityOfServiceMode.valueOf(mode)); + } + else + { + mode = null; + } + } + + public void setSelector(String selector) + { + bridge.setSelector(selector); + } + + public void setSourcePassword(String pwd) + { + bridge.setSourcePassword(pwd); + } + + public void setSourceUsername(String name) + { + bridge.setSourceUsername(name); + } + + public void setSubscriptionName(String subname) + { + bridge.setSubscriptionName(subname); + } + + public void setTargetPassword(String pwd) + { + bridge.setTargetPassword(pwd); + } + + public void setTargetUsername(String name) + { + bridge.setTargetUsername(name); + } + + public void setTransactionManagerLocatorClass(String transactionManagerLocatorClass) + { + bridge.setTransactionManagerLocatorClass(transactionManagerLocatorClass); + } + + public void setTransactionManagerLocatorMethod(String transactionManagerLocatorMethod) + { + bridge.setTransactionManagerLocatorMethod(transactionManagerLocatorMethod); + } + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java 2009-11-02 15:32:07 UTC (rev 8183) +++ trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java 2009-11-02 16:03:05 UTC (rev 8184) @@ -27,13 +27,15 @@ import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; -import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.Topic; import javax.jms.XAConnection; import javax.jms.XAConnectionFactory; import javax.jms.XASession; +import javax.management.MBeanServer; +import javax.management.ObjectName; +import javax.management.StandardMBean; import javax.transaction.Transaction; import javax.transaction.TransactionManager; import javax.transaction.xa.XAResource; @@ -44,6 +46,7 @@ import org.hornetq.jms.bridge.ConnectionFactoryFactory; import org.hornetq.jms.bridge.DestinationFactory; import org.hornetq.jms.bridge.JMSBridge; +import org.hornetq.jms.bridge.JMSBridgeControl; import org.hornetq.jms.bridge.QualityOfServiceMode; import org.hornetq.jms.client.HornetQMessage; import org.hornetq.jms.client.HornetQSession; @@ -150,6 +153,10 @@ private String transactionManagerLocatorClass = "org.hornetq.integration.jboss.tm.JBoss5TransactionManagerLocator"; private String transactionManagerLocatorMethod = "getTm"; + + private MBeanServer mbeanServer; + + private ObjectName objectName; private static final int FORWARD_MODE_XA = 0; @@ -164,8 +171,24 @@ { this.messages = new LinkedList<Message>(); } - + public JMSBridgeImpl(ConnectionFactoryFactory sourceCff, ConnectionFactoryFactory targetCff, + DestinationFactory sourceDestinationFactory, DestinationFactory targetDestinationFactory, + String sourceUsername, String sourcePassword, + String targetUsername, String targetPassword, + String selector, long failureRetryInterval, + int maxRetries, + QualityOfServiceMode qosMode, + int maxBatchSize, long maxBatchTime, + String subName, String clientID, + boolean addMessageIDInHeader) { + + this(sourceCff, targetCff, sourceDestinationFactory, targetDestinationFactory, sourceUsername, + sourcePassword, targetUsername, targetPassword, selector, failureRetryInterval, maxRetries, + qosMode, maxBatchSize, maxBatchTime, subName, clientID, addMessageIDInHeader, null, null); + } + + public JMSBridgeImpl(ConnectionFactoryFactory sourceCff, ConnectionFactoryFactory targetCff, DestinationFactory sourceDestinationFactory, DestinationFactory targetDestinationFactory, String sourceUsername, String sourcePassword, String targetUsername, String targetPassword, @@ -174,7 +197,9 @@ QualityOfServiceMode qosMode, int maxBatchSize, long maxBatchTime, String subName, String clientID, - boolean addMessageIDInHeader) + boolean addMessageIDInHeader, + MBeanServer mbeanServer, + String objectName) { this(); @@ -211,8 +236,32 @@ this.clientID = clientID; this.addMessageIDInHeader = addMessageIDInHeader; - - checkParams(); + + checkParams(); + + if(mbeanServer != null) + { + if(objectName != null) + { + this.mbeanServer = mbeanServer; + + try + { + JMSBridgeControlImpl controlBean = new JMSBridgeControlImpl(this); + this.objectName = ObjectName.getInstance(objectName); + StandardMBean mbean = new StandardMBean(controlBean, JMSBridgeControl.class); + mbeanServer.registerMBean(mbean, this.objectName); + log.debug("Registered JMSBridge instance as: " + this.objectName.getCanonicalName()); + } + catch (Exception e) + { + throw new IllegalStateException("Failed to register JMSBridge MBean", e); + } + } + else { + throw new IllegalArgumentException("objectName is required when specifying an MBeanServer"); + } + } if (trace) { @@ -382,6 +431,21 @@ { return started; } + + public void destroy() + { + if (mbeanServer != null && objectName != null) + { + try + { + mbeanServer.unregisterMBean(objectName); + } + catch (Exception e) + { + log.warn("Failed to unregisted JMS Bridge " + objectName); + } + } + } // JMSBridge implementation ------------------------------------------------------------ Modified: trunk/tests/src/org/hornetq/tests/integration/jms/bridge/JMSBridgeTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/bridge/JMSBridgeTest.java 2009-11-02 15:32:07 UTC (rev 8183) +++ trunk/tests/src/org/hornetq/tests/integration/jms/bridge/JMSBridgeTest.java 2009-11-02 16:03:05 UTC (rev 8184) @@ -12,6 +12,7 @@ */ package org.hornetq.tests.integration.jms.bridge; +import java.lang.management.ManagementFactory; import java.util.ArrayList; import java.util.Iterator; import java.util.List; @@ -23,6 +24,8 @@ import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; +import javax.management.MBeanServer; +import javax.management.ObjectName; import javax.transaction.Transaction; import javax.transaction.TransactionManager; @@ -1600,6 +1603,26 @@ } } + public void testMBeanServer() throws Exception { + + MBeanServer mbeanServer = ManagementFactory.getPlatformMBeanServer(); + ObjectName objectName = new ObjectName("example.jmsbridge:service=JMSBridge"); + + JMSBridgeImpl bridge = new JMSBridgeImpl(cff0, cff0, sourceQueueFactory, localTargetQueueFactory, + null, null, null, null, + null, 5000, 10, QualityOfServiceMode.AT_MOST_ONCE, + 1, -1, + null, null, false, + mbeanServer, + objectName.getCanonicalName()); + + assertTrue(mbeanServer.isRegistered(objectName)); + + bridge.destroy(); + + assertFalse(mbeanServer.isRegistered(objectName)); + } + public TransactionManager getNewTm() { return newTransactionManager();

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8183 - in trunk/docs/user-manual: en and 1 other directory.

by do-not-reply＠jboss.org

Author: jmesnil Date: 2009-11-02 10:32:07 -0500 (Mon, 02 Nov 2009) New Revision: 8183 Added: trunk/docs/user-manual/en/jms-bridge.xml Modified: trunk/docs/user-manual/en/appserver-integration.xml trunk/docs/user-manual/en/examples.xml trunk/docs/user-manual/en/master.xml trunk/docs/user-manual/user-manual.xpr Log: JMS Bridge documentation * moved JMS bridge section outside of the appserver integration chapter * added ref to the jms/jms-bridge example Modified: trunk/docs/user-manual/en/appserver-integration.xml =================================================================== --- trunk/docs/user-manual/en/appserver-integration.xml 2009-11-02 15:07:01 UTC (rev 8182) +++ trunk/docs/user-manual/en/appserver-integration.xml 2009-11-02 15:32:07 UTC (rev 8183) @@ -642,387 +642,7 @@ url="http://www.jboss.org/file-access/default/members/jbossas/freezone/docs/Cl..." >JBoss Application Server clustering documentation</ulink></para> </section> - <section id="jms-bridge"> - <title>The JMS Bridge</title> - <para>HornetQ includes a fully functional message bridge.</para> - <para>The function of the bridge is to consume messages from a source queue or topic, and - send them to a target queue or topic, typically on a different server.</para> - <para>The source and target servers do not have to be in the same cluster which makes - bridging suitable for reliably sending messages from one cluster to another, for - instance across a WAN, and where the connection may be unreliable.</para> - <para>A bridge is deployed inside a JBoss AS instance. The instance can be the same instance - as either the source or target server. Or could be on a third, separate JBoss AS - instance.</para> - <para>The bridge can also be used to bridge messages from other non HornetQ JMS servers, as - long as they are JMS 1.1 compliant.<note><para>Don't confuse a JMS bridge with a core - bridge. A JMS bridge can be used to bridge any two JMS 1.1 compliant JMS - providers and uses the JMS API. A core bridge (described in <xref - linkend="core-bridges"/>) is used to bridge any two HornetQ instances and - uses the core API. Always use a core bridge if you can in preference to a JMS - bridge. The core bridge will typically provide better performance than a JMS - bridge. Also the core bridge can provide <emphasis>once and only once</emphasis> - delivery guarantees without using XA.</para></note></para> - <para>The bridge has built-in resilience to failure so if the source or target server - connection is lost, e.g. due to network failure, the bridge will retry connecting to the - source and/or target until they come back online. When it comes back online it will - resume operation as normal.</para> - <para>The bridge can be configured with an optional JMS selector, so it will only consume - messages matching that JMS selector</para> - <para>It can be configured to consume from a queue or a topic. When it consumes from a topic - it can be configured to consume using a non durable or durable subscription</para> - <para>The bridge is deployed by the JBoss Micro Container via a beans configuration file. - This would typically be deployed inside the JBoss Application Server and the following - example shows an example of a beans file that bridges 2 destinations which are actually - on the same server. </para> - <programlisting><?xml version="1.0" encoding="UTF-8"?> -<deployment xmlns="urn:jboss:bean-deployer:2.0"> - - <bean name="JMSBridge" class="org.hornetq.jms.bridge.impl.JMSBridgeImpl"> -  - <depends>HornetQServer</depends> - <constructor> -  - <parameter> - <inject bean="SourceCFF"/> - </parameter> -  - <parameter> - <inject bean="TargetCFF"/> - </parameter> -  - <parameter> - <inject bean="SourceDestinationFactory"/> - </parameter> -  - <parameter> - <inject bean="TargetDestinationFactory"/> - </parameter> -  - <parameter><null /></parameter> -  - <parameter><null /></parameter> -  - <parameter><null /></parameter> -  - <parameter><null /></parameter> -  - <parameter><null /></parameter> -  - <parameter>5000</parameter> -  - <parameter>10</parameter> -  - <parameter>ONCE_AND_ONLY_ONCE</parameter> -  - <parameter>1</parameter> -  - <parameter>-1</parameter> -  - <parameter><null /></parameter> -  - <parameter><null /></parameter> -  - <parameter>true</parameter> - </constructor> - <property name="transactionManager"> - <inject bean="RealTransactionManager"/> - </property> - </bean> - -  - <bean name="SourceCFF" - class="org.hornetq.jms.bridge.impl.JNDIConnectionFactoryFactory"> - <constructor> - <parameter> - <inject bean="JNDI" /> - </parameter> - <parameter>/ConnectionFactory</parameter> - </constructor> - </bean> - -  - <bean name="TargetCFF" - class="org.hornetq.jms.bridge.impl.JNDIConnectionFactoryFactory"> - <constructor> - <parameter> - <inject bean="JNDI" /> - </parameter> - <parameter>/ConnectionFactory</parameter> - </constructor> - </bean> - -  - <bean name="SourceDestinationFactory" - class="org.hornetq.jms.bridge.impl.JNDIDestinationFactory"> - <constructor> - <parameter> - <inject bean="JNDI" /> - </parameter> - <parameter>/queue/source</parameter> - </constructor> - </bean> - -  - <bean name="TargetDestinationFactory" - class="org.hornetq.jms.bridge.impl.JNDIDestinationFactory"> - <constructor> - <parameter> - <inject bean="JNDI" /> - </parameter> - <parameter>/queue/target</parameter> - </constructor> - </bean> - -  -  - <bean name="JNDI" class="java.util.Hashtable"> - <constructor class="java.util.Map"> - <map class="java.util.Hashtable" keyClass="String" - valueClass="String"> - <entry> - <key>java.naming.factory.initial</key> - <value>org.jnp.interfaces.NamingContextFactory</value> - </entry> - <entry> - <key>java.naming.provider.url</key> - <value>jnp://localhost:1099</value> - </entry> - <entry> - <key>java.naming.factory.url.pkgs</key> - <value>org.jboss.naming:org.jnp.interfaces"</value> - </entry> - </map> - </constructor> - </bean> - -</deployment></programlisting> - <section> - <title>JMS Bridge Parameters</title> - <para>The main bean deployed is the <literal>JMSBridge</literal> bean. The bean is - configurable by the parameters passed to its constructor.</para> - <note> - <para>To let a parameter be unspecified (for example, if the authentication is - anonymous or no message selector is provided), use <literal><null - /></literal> for the unspecified parameter value.</para> - </note> - <itemizedlist> - <listitem> - <para>Source Connection Factory Factory</para> - <para>This injects the <literal>SourceCFF</literal> bean (also defined in the - beans file). This bean is used to create the <emphasis>source</emphasis> - <literal>ConnectionFactory</literal> - </para> - </listitem> - <listitem> - <para>Target Connection Factory Factory</para> - <para>This injects the <literal>TargetCFF</literal> bean (also defined in the - beans file). This bean is used to create the <emphasis>target</emphasis> - <literal>ConnectionFactory</literal> - </para> - </listitem> - <listitem> - <para>Source Destination Factory Factory</para> - <para>This injects the <literal>SourceDestinationFactory</literal> bean (also - defined in the beans file). This bean is used to create the - <emphasis>source</emphasis> - <literal>Destination</literal> - </para> - </listitem> - <listitem> - <para>Target Destination Factory Factory</para> - <para>This injects the <literal>TargetDestinationFactory</literal> bean (also - defined in the beans file). This bean is used to create the - <emphasis>target</emphasis> - <literal>Destination</literal> - </para> - </listitem> - <listitem> - <para>Source User Name</para> - <para>this parameter is the username for creating the - <emphasis>source</emphasis> connection</para> - </listitem> - <listitem> - <para>Source Password</para> - <para>this parameter is the parameter for creating the - <emphasis>source</emphasis> connection</para> - </listitem> - <listitem> - <para>Target User Name</para> - <para>this parameter is the username for creating the - <emphasis>target</emphasis> connection</para> - </listitem> - <listitem> - <para>Target Password</para> - <para>this parameter is the password for creating the - <emphasis>target</emphasis> connection</para> - </listitem> - <listitem> - <para>Selector</para> - <para>This represents a JMS selector expression used for consuming messages from - the source destination. Only messages that match the selector expression - will be bridged from the source to the target destination</para> - <para> - <note>Ut is always more efficient to apply selectors on source topic - subscriptions to source queue consumers</note> - </para> - <para>The selector expression must follow the <ulink - url="http://java.sun.com/j2ee/1.4/docs/api/javax/jms/Message.html">JMS - selector syntax</ulink></para> - </listitem> - <listitem> - <para>Failure Retry Interval</para> - <para>This represents the amount of time in ms to wait between trying to - recreate connections to the source or target servers when the bridge has - detected they have failed</para> - </listitem> - <listitem> - <para>Max Retries</para> - <para>This represents the number of times to attempt to recreate connections to - the source or target servers when the bridge has detected they have failed. - The bridge will give up after trying this number of times. <literal - >-1</literal> represents 'try forever'</para> - </listitem> - <listitem> - <para>Quality Of Service</para> - <para>This parameter represents the desired quality of service mode</para> - <para>Possible values are:</para> - <itemizedlist> - <listitem> - <para><literal>AT_MOST_ONCE</literal></para> - </listitem> - <listitem> - <para><literal>DUPLICATES_OK</literal></para> - </listitem> - <listitem> - <para><literal>ONCE_AND_ONLY_ONCE</literal></para> - </listitem> - </itemizedlist> - <para>See <xref linkend="quality-of-service"/> for a explanation of these - modes.</para> - </listitem> - <listitem> - <para>Max Batch Size</para> - <para>This represents the maximum number of messages to consume from the source - destination before sending them in a batch to the target destination. Its - value must <literal>>= 1</literal> - </para> - </listitem> - <listitem> - <para>Max Batch Time</para> - <para>This represents the maximum number of milliseconds to wait before sending - a batch to target, even if the number of messages consumed has not reached - <literal>MaxBatchSize</literal>. Its value must be <literal>-1</literal> - to represent 'wait forever', or <literal>>= 1</literal> to specify an actual - time </para> - </listitem> - <listitem> - <para>Subscription Name</para> - <para>If the source destination represents a topic, and you want to consume from - the topic using a durable subscription then this parameter represents the - durable subscription name</para> - </listitem> - <listitem> - <para>Client ID</para> - <para>If the source destination represents a topic, and you want to consume from - the topic using a durable subscription then this attribute represents the - the JMS client ID to use when creating/looking up the durable - subscription</para> - </listitem> - <listitem> - <para>Add MessageID In Header</para> - <para>If <literal>true</literal>, then the original message's message ID will be - appended in the message sent to the destination in the header <literal - >HORNETQ_BRIDGE_MSG_ID_LIST</literal>. If the message is bridged more - than once, each message ID will be appended. This enables a distributed - request-response pattern to be used</para> - <note> - <para>when you receive the message you can send back a response using the - correlation id of the first message id, so when the original sender gets - it back it will be able to correlate it. </para> - </note> - </listitem> - </itemizedlist> - </section> - <section> - <title>Source and Target Connection Factories</title> - <para>The source and target connection factory factories are used to create the - connection factory used to create the connection for the source or target - server.</para> - <para>The configuration example above uses the default implementation provided by - HornetQ that looks up the connection factory using JNDI. For other Application - Servers or JMS providers a new implementation may have to be provided. This can - easily be done by implementing the interface <literal - >org.hornetq.jms.bridge.ConnectionFactoryFactory</literal>.</para> - </section> - <section> - <title>Source and Target Destination Factories</title> - <para>Again, similarly, these are used to create or lookup up the destinations.</para> - <para>In the configuration example above, we have used the default provided by HornetQ - that looks up the destination using JNDI.</para> - <para>A new implementation can be provided by implementing <literal - >org.hornetq.jms.bridge.DestinationFactory</literal> interface.</para> - </section> - <section id="quality-of-service"> - <title>Quality Of Service</title> - <para>The quality of service modes used by the bridge are described here in more - detail.</para> - <section> - <title>AT_MOST_ONCE</title> - <para>With this QoS mode messages will reach the destination from the source at most - once. The messages are consumed from the source and acknowledged before sending - to the destination. Therefore there is a possibility that if failure occurs - between removing them from the source and them arriving at the destination they - could be lost. Hence delivery will occur at most once.</para> - <para>This mode is available for both persistent and non persistent messages.</para> - </section> - <section> - <title>DUPLICATES_OK</title> - <para>With this QoS mode, the messages are consumed from the source and then - acknowledged after they have been successfully sent to the destination. - Therefore there is a possibility that if failure occurs after sending to the - destination but before acknowledging them, they could be sent again when the - system recovers. I.e. the destination might receive duplicates after a - failure.</para> - <para>This mode is available for both persistent and non persistent messages.</para> - </section> - <section> - <title>ONCE_AND_ONLY_ONCE</title> - <para>This QoS mode ensures messages will reach the destination from the source once - and only once. (Sometimes this mode is known as "exactly once"). If both the - source and the destination are on the same HornetQ server instance then this can - be achieved by sending and acknowledging the messages in the same local - transaction. If the source and destination are on different servers this is - achieved by enlisting the sending and consuming sessions in a JTA transaction. - The JTA transaction is controlled by JBoss Transactions JTA * implementation - which is a fully recovering transaction manager, thus providing a very high - degree of durability. If JTA is required then both supplied connection factories - need to be XAConnectionFactory implementations. This is likely to be the slowest - mode since it requires extra persistence for the transaction logging.</para> - <para>This mode is only available for persistent messages.</para> - <note> - <para>For a specific application it may possible to provide once and only once - semantics without using the ONCE_AND_ONLY_ONCE QoS level. This can be done - by using the DUPLICATES_OK mode and then checking for duplicates at the - destination and discarding them. Some JMS servers provide automatic - duplicate message detection functionality, or this may be possible to - implement on the application level by maintaining a cache of received - message ids on disk and comparing received messages to them. The cache would - only be valid for a certain period of time so this approach is not as - watertight as using ONCE_AND_ONLY_ONCE but may be a good choice depending on - your specific application.</para> - </note> - </section> - <section> - <title>Example</title> - <para>Please see <xref linkend="jms-bridge-example"/> which shows how to configure - and use a JMS Bridge to send messages to the source destination and consume them - from the target destination.</para> - </section> - </section> - </section> <section id="xa-recovery"> <title>XA Recovery</title> <para><emphasis>XA recovery</emphasis> deals with system or application failures to ensure Modified: trunk/docs/user-manual/en/examples.xml =================================================================== --- trunk/docs/user-manual/en/examples.xml 2009-11-02 15:07:01 UTC (rev 8182) +++ trunk/docs/user-manual/en/examples.xml 2009-11-02 15:32:07 UTC (rev 8183) @@ -194,6 +194,11 @@ for security. HornetQ can leverage JAAS to delegate user authentication and authorization to existing security infrastructure.</para> </section> + <section id="examples.jms.jms-bridge"> + <title>JMS Bridge</title> + <para>The <literal>jms-brige</literal> example shows how to setup a bridge + between two standalone HornetQ servers.</para> + </section> <section id="examples.jmx"> <title>JMX Management</title> <para>The <literal>jmx</literal> example shows how to manage HornetQ using JMX.</para> @@ -512,7 +517,7 @@ <para>This example demonstrates how to configure several properties on the HornetQ JCA resource adaptor.</para> </section> - <section id="jms-bridge-example"> + <section id="examples.javaee.jms-bridge"> <title>JMS Bridge</title> <para>An example demonstrating the use of the HornetQ JMS bridge.</para> </section> Added: trunk/docs/user-manual/en/jms-bridge.xml =================================================================== --- trunk/docs/user-manual/en/jms-bridge.xml (rev 0) +++ trunk/docs/user-manual/en/jms-bridge.xml 2009-11-02 15:32:07 UTC (rev 8183) @@ -0,0 +1,402 @@ +<?xml version="1.0" encoding="UTF-8"?> + + + + + + + + + + + + + + + + + +<chapter id="jms-bridge"> + <section> + <title>The JMS Bridge</title> + <para>HornetQ includes a fully functional JMS message bridge.</para> + <para>The function of the bridge is to consume messages from a source queue or topic, and + send them to a target queue or topic, typically on a different server.</para> + <para>The source and target servers do not have to be in the same cluster which makes + bridging suitable for reliably sending messages from one cluster to another, for + instance across a WAN, and where the connection may be unreliable.</para> + <para>A bridge can be deployed as a standalone application, with HornetQ standalone server or inside a JBoss AS + instance. The source and the target can be located in the same virtual machine or another one.</para> + <para>The bridge can also be used to bridge messages from other non HornetQ JMS servers, as + long as they are JMS 1.1 compliant.<note><para>Do not confuse a JMS bridge with a core + bridge. A JMS bridge can be used to bridge any two JMS 1.1 compliant JMS + providers and uses the JMS API. A core bridge (described in <xref + linkend="core-bridges"/>) is used to bridge any two HornetQ instances and + uses the core API. Always use a core bridge if you can in preference to a JMS + bridge. The core bridge will typically provide better performance than a JMS + bridge. Also the core bridge can provide <emphasis>once and only once</emphasis> + delivery guarantees without using XA.</para></note></para> + <para>The bridge has built-in resilience to failure so if the source or target server + connection is lost, e.g. due to network failure, the bridge will retry connecting to the + source and/or target until they come back online. When it comes back online it will + resume operation as normal.</para> + <para>The bridge can be configured with an optional JMS selector, so it will only consume + messages matching that JMS selector</para> + <para>It can be configured to consume from a queue or a topic. When it consumes from a topic + it can be configured to consume using a non durable or durable subscription</para> + <para>Typically, the bridge is deployed by the JBoss Micro Container via a beans configuration file. + This would typically be deployed inside the JBoss Application Server and the following + example shows an example of a beans file that bridges 2 destinations which are actually + on the same server. </para> + <programlisting><?xml version="1.0" encoding="UTF-8"?> + +<deployment xmlns="urn:jboss:bean-deployer:2.0"> + + <bean name="JMSBridge" class="org.hornetq.jms.bridge.impl.JMSBridgeImpl"> +  + <depends>HornetQServer</depends> + <constructor> +  + <parameter> + <inject bean="SourceCFF"/> + </parameter> +  + <parameter> + <inject bean="TargetCFF"/> + </parameter> +  + <parameter> + <inject bean="SourceDestinationFactory"/> + </parameter> +  + <parameter> + <inject bean="TargetDestinationFactory"/> + </parameter> +  + <parameter><null /></parameter> +  + <parameter><null /></parameter> +  + <parameter><null /></parameter> +  + <parameter><null /></parameter> +  + <parameter><null /></parameter> +  + <parameter>5000</parameter> +  + <parameter>10</parameter> +  + <parameter>ONCE_AND_ONLY_ONCE</parameter> +  + <parameter>1</parameter> +  + <parameter>-1</parameter> +  + <parameter><null /></parameter> +  + <parameter><null /></parameter> +  + <parameter>true</parameter> + </constructor> + <property name="transactionManager"> + <inject bean="RealTransactionManager"/> + </property> + </bean> + +  + <bean name="SourceCFF" + class="org.hornetq.jms.bridge.impl.JNDIConnectionFactoryFactory"> + <constructor> + <parameter> + <inject bean="JNDI" /> + </parameter> + <parameter>/ConnectionFactory</parameter> + </constructor> + </bean> + +  + <bean name="TargetCFF" + class="org.hornetq.jms.bridge.impl.JNDIConnectionFactoryFactory"> + <constructor> + <parameter> + <inject bean="JNDI" /> + </parameter> + <parameter>/ConnectionFactory</parameter> + </constructor> + </bean> + +  + <bean name="SourceDestinationFactory" + class="org.hornetq.jms.bridge.impl.JNDIDestinationFactory"> + <constructor> + <parameter> + <inject bean="JNDI" /> + </parameter> + <parameter>/queue/source</parameter> + </constructor> + </bean> + +  + <bean name="TargetDestinationFactory" + class="org.hornetq.jms.bridge.impl.JNDIDestinationFactory"> + <constructor> + <parameter> + <inject bean="JNDI" /> + </parameter> + <parameter>/queue/target</parameter> + </constructor> + </bean> + +  +  + <bean name="JNDI" class="java.util.Hashtable"> + <constructor class="java.util.Map"> + <map class="java.util.Hashtable" keyClass="String" + valueClass="String"> + <entry> + <key>java.naming.factory.initial</key> + <value>org.jnp.interfaces.NamingContextFactory</value> + </entry> + <entry> + <key>java.naming.provider.url</key> + <value>jnp://localhost:1099</value> + </entry> + <entry> + <key>java.naming.factory.url.pkgs</key> + <value>org.jboss.naming:org.jnp.interfaces"</value> + </entry> + </map> + </constructor> + </bean> + +</deployment></programlisting> + <section> + <title>JMS Bridge Parameters</title> + <para>The main bean deployed is the <literal>JMSBridge</literal> bean. The bean is + configurable by the parameters passed to its constructor.</para> + <note> + <para>To let a parameter be unspecified (for example, if the authentication is + anonymous or no message selector is provided), use <literal><null + /></literal> for the unspecified parameter value.</para> + </note> + <itemizedlist> + <listitem> + <para>Source Connection Factory Factory</para> + <para>This injects the <literal>SourceCFF</literal> bean (also defined in the + beans file). This bean is used to create the <emphasis>source</emphasis> + <literal>ConnectionFactory</literal> + </para> + </listitem> + <listitem> + <para>Target Connection Factory Factory</para> + <para>This injects the <literal>TargetCFF</literal> bean (also defined in the + beans file). This bean is used to create the <emphasis>target</emphasis> + <literal>ConnectionFactory</literal> + </para> + </listitem> + <listitem> + <para>Source Destination Factory Factory</para> + <para>This injects the <literal>SourceDestinationFactory</literal> bean (also + defined in the beans file). This bean is used to create the + <emphasis>source</emphasis> + <literal>Destination</literal> + </para> + </listitem> + <listitem> + <para>Target Destination Factory Factory</para> + <para>This injects the <literal>TargetDestinationFactory</literal> bean (also + defined in the beans file). This bean is used to create the + <emphasis>target</emphasis> + <literal>Destination</literal> + </para> + </listitem> + <listitem> + <para>Source User Name</para> + <para>this parameter is the username for creating the + <emphasis>source</emphasis> connection</para> + </listitem> + <listitem> + <para>Source Password</para> + <para>this parameter is the parameter for creating the + <emphasis>source</emphasis> connection</para> + </listitem> + <listitem> + <para>Target User Name</para> + <para>this parameter is the username for creating the + <emphasis>target</emphasis> connection</para> + </listitem> + <listitem> + <para>Target Password</para> + <para>this parameter is the password for creating the + <emphasis>target</emphasis> connection</para> + </listitem> + <listitem> + <para>Selector</para> + <para>This represents a JMS selector expression used for consuming messages from + the source destination. Only messages that match the selector expression + will be bridged from the source to the target destination</para> + <para> + <note>Ut is always more efficient to apply selectors on source topic + subscriptions to source queue consumers</note> + </para> + <para>The selector expression must follow the <ulink + url="http://java.sun.com/j2ee/1.4/docs/api/javax/jms/Message.html">JMS + selector syntax</ulink></para> + </listitem> + <listitem> + <para>Failure Retry Interval</para> + <para>This represents the amount of time in ms to wait between trying to + recreate connections to the source or target servers when the bridge has + detected they have failed</para> + </listitem> + <listitem> + <para>Max Retries</para> + <para>This represents the number of times to attempt to recreate connections to + the source or target servers when the bridge has detected they have failed. + The bridge will give up after trying this number of times. <literal + >-1</literal> represents 'try forever'</para> + </listitem> + <listitem> + <para>Quality Of Service</para> + <para>This parameter represents the desired quality of service mode</para> + <para>Possible values are:</para> + <itemizedlist> + <listitem> + <para><literal>AT_MOST_ONCE</literal></para> + </listitem> + <listitem> + <para><literal>DUPLICATES_OK</literal></para> + </listitem> + <listitem> + <para><literal>ONCE_AND_ONLY_ONCE</literal></para> + </listitem> + </itemizedlist> + <para>See <xref linkend="quality-of-service"/> for a explanation of these + modes.</para> + </listitem> + <listitem> + <para>Max Batch Size</para> + <para>This represents the maximum number of messages to consume from the source + destination before sending them in a batch to the target destination. Its + value must <literal>>= 1</literal> + </para> + </listitem> + <listitem> + <para>Max Batch Time</para> + <para>This represents the maximum number of milliseconds to wait before sending + a batch to target, even if the number of messages consumed has not reached + <literal>MaxBatchSize</literal>. Its value must be <literal>-1</literal> + to represent 'wait forever', or <literal>>= 1</literal> to specify an actual + time </para> + </listitem> + <listitem> + <para>Subscription Name</para> + <para>If the source destination represents a topic, and you want to consume from + the topic using a durable subscription then this parameter represents the + durable subscription name</para> + </listitem> + <listitem> + <para>Client ID</para> + <para>If the source destination represents a topic, and you want to consume from + the topic using a durable subscription then this attribute represents the + the JMS client ID to use when creating/looking up the durable + subscription</para> + </listitem> + <listitem> + <para>Add MessageID In Header</para> + <para>If <literal>true</literal>, then the original message's message ID will be + appended in the message sent to the destination in the header <literal + >HORNETQ_BRIDGE_MSG_ID_LIST</literal>. If the message is bridged more + than once, each message ID will be appended. This enables a distributed + request-response pattern to be used</para> + <note> + <para>when you receive the message you can send back a response using the + correlation id of the first message id, so when the original sender gets + it back it will be able to correlate it. </para> + </note> + </listitem> + </itemizedlist> + </section> + <section> + <title>Source and Target Connection Factories</title> + <para>The source and target connection factory factories are used to create the + connection factory used to create the connection for the source or target + server.</para> + <para>The configuration example above uses the default implementation provided by + HornetQ that looks up the connection factory using JNDI. For other Application + Servers or JMS providers a new implementation may have to be provided. This can + easily be done by implementing the interface <literal + >org.hornetq.jms.bridge.ConnectionFactoryFactory</literal>.</para> + </section> + <section> + <title>Source and Target Destination Factories</title> + <para>Again, similarly, these are used to create or lookup up the destinations.</para> + <para>In the configuration example above, we have used the default provided by HornetQ + that looks up the destination using JNDI.</para> + <para>A new implementation can be provided by implementing <literal + >org.hornetq.jms.bridge.DestinationFactory</literal> interface.</para> + </section> + <section id="quality-of-service"> + <title>Quality Of Service</title> + <para>The quality of service modes used by the bridge are described here in more + detail.</para> + <section> + <title>AT_MOST_ONCE</title> + <para>With this QoS mode messages will reach the destination from the source at most + once. The messages are consumed from the source and acknowledged before sending + to the destination. Therefore there is a possibility that if failure occurs + between removing them from the source and them arriving at the destination they + could be lost. Hence delivery will occur at most once.</para> + <para>This mode is available for both persistent and non persistent messages.</para> + </section> + <section> + <title>DUPLICATES_OK</title> + <para>With this QoS mode, the messages are consumed from the source and then + acknowledged after they have been successfully sent to the destination. + Therefore there is a possibility that if failure occurs after sending to the + destination but before acknowledging them, they could be sent again when the + system recovers. I.e. the destination might receive duplicates after a + failure.</para> + <para>This mode is available for both persistent and non persistent messages.</para> + </section> + <section> + <title>ONCE_AND_ONLY_ONCE</title> + <para>This QoS mode ensures messages will reach the destination from the source once + and only once. (Sometimes this mode is known as "exactly once"). If both the + source and the destination are on the same HornetQ server instance then this can + be achieved by sending and acknowledging the messages in the same local + transaction. If the source and destination are on different servers this is + achieved by enlisting the sending and consuming sessions in a JTA transaction. + The JTA transaction is controlled by JBoss Transactions JTA * implementation + which is a fully recovering transaction manager, thus providing a very high + degree of durability. If JTA is required then both supplied connection factories + need to be XAConnectionFactory implementations. This is likely to be the slowest + mode since it requires extra persistence for the transaction logging.</para> + <para>This mode is only available for persistent messages.</para> + <note> + <para>For a specific application it may possible to provide once and only once + semantics without using the ONCE_AND_ONLY_ONCE QoS level. This can be done + by using the DUPLICATES_OK mode and then checking for duplicates at the + destination and discarding them. Some JMS servers provide automatic + duplicate message detection functionality, or this may be possible to + implement on the application level by maintaining a cache of received + message ids on disk and comparing received messages to them. The cache would + only be valid for a certain period of time so this approach is not as + watertight as using ONCE_AND_ONLY_ONCE but may be a good choice depending on + your specific application.</para> + </note> + </section> + <section> + <title>Examples</title> + <para>Please see <xref linkend="examples.javaee.jms-bridge"/> which shows how to configure + and use a JMS Bridge with JBoss AS to send messages to the source destination and consume them + from the target destination.</para> + <para>Please see <xref linkend="examples.jms.jms-bridge"/> which shows how to configure + and use a JMS Bridge between two standalone HornetQ servers.</para> + </section> + </section> + </section> +</chapter> Modified: trunk/docs/user-manual/en/master.xml =================================================================== --- trunk/docs/user-manual/en/master.xml 2009-11-02 15:07:01 UTC (rev 8182) +++ trunk/docs/user-manual/en/master.xml 2009-11-02 15:32:07 UTC (rev 8183) @@ -40,6 +40,7 @@ <!ENTITY transaction-config SYSTEM "transaction-config.xml"> <!ENTITY intercepting-operations SYSTEM "intercepting-operations.xml"> <!ENTITY interoperability SYSTEM "interoperability.xml"> + <!ENTITY jms-bridge SYSTEM "jms-bridge.xml"> <!ENTITY jms-core-mapping SYSTEM "jms-core-mapping.xml"> <!ENTITY large-messages SYSTEM "large-messages.xml"> <!ENTITY last-value-queues SYSTEM "last-value-queues.xml"> @@ -109,6 +110,7 @@ &management; &security; &appserver-integration; + &jms-bridge; &client-reconnection; &diverts; &core-bridges; Modified: trunk/docs/user-manual/user-manual.xpr =================================================================== --- trunk/docs/user-manual/user-manual.xpr 2009-11-02 15:07:01 UTC (rev 8182) +++ trunk/docs/user-manual/user-manual.xpr 2009-11-02 15:32:07 UTC (rev 8183) @@ -32,6 +32,7 @@ <file name="en/ha.xml"/> <file name="en/intercepting-operations.xml"/> <file name="en/interoperability.xml"/> + <file name="en/jms-bridge.xml"/> <file name="en/jms-core-mapping.xml"/> <file name="en/large-messages.xml"/> <file name="en/last-value-queues.xml"/>

16 years, 1 month

1
0
0 / 0

JBoss hornetq SVN: r8182 - trunk/examples/common.

by do-not-reply＠jboss.org

Author: jmesnil Date: 2009-11-02 10:07:01 -0500 (Mon, 02 Nov 2009) New Revision: 8182 Modified: trunk/examples/common/build.xml Log: automated run of examples Added clustered-standalone to the list of examples to run manually Modified: trunk/examples/common/build.xml =================================================================== --- trunk/examples/common/build.xml 2009-11-02 15:04:10 UTC (rev 8181) +++ trunk/examples/common/build.xml 2009-11-02 15:07:01 UTC (rev 8182) @@ -199,12 +199,13 @@ <exclude name="common/build.xml"/> <exclude name="core/perf/build.xml"/> <exclude name="jms/applet/build.xml"/> + <exclude name="jms/clustered-standalone/build.xml"/> <exclude name="jms/jms-bridge/build.xml"/> <exclude name="jms/perf/build.xml"/> </fileset> </subant> - <echo message="Do not forget to run jms/applet & jms-bridge examples manually!" /> + <echo message="Do not forget to run jms/applet, jms/clustered-standalone & jms-bridge examples manually!" /> </target> </project>

16 years, 1 month

1
0
0 / 0

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

hornetq-commits November 2009