Author: clebert.suconic(a)jboss.com Date: 2012-01-11 16:25:18 -0500 (Wed, 11 Jan 2012) New Revision: 12008 Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/SimpleString.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/RemotingService.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/clientcrash/ClientTestBase.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/LinkedListTest.java branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/SimpleStringTest.java Log: merging changes from EAP Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/SimpleString.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/SimpleString.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/api/core/SimpleString.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -25,7 +25,7 @@ * this minimises expensive copying between String objects. * * This object is used heavily throughout HornetQ for performance reasons. - * + * * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * */ @@ -116,7 +116,7 @@ } pos <<= 1; - return (char)(data[pos] | data[pos + 1] << 8); + return (char)((data[pos] & 0xFF) | (data[pos + 1] << 8) & 0xFF00); } public CharSequence subSequence(final int start, final int end) @@ -215,7 +215,7 @@ { return true; } - + if (other instanceof SimpleString) { SimpleString s = (SimpleString)other; @@ -258,9 +258,8 @@ } /** - * splits this SimpleString into an array of SimpleString using the char param as the delimeter. - * - * i.e. "a.b" would return "a" and "b" if . was the delimeter + * Splits this SimpleString into an array of SimpleString using the char param as the delimiter. + * i.e. "a.b" would return "a" and "b" if . was the delimiter * @param delim */ public SimpleString[] split(final char delim) @@ -272,11 +271,13 @@ else { List<SimpleString> all = new ArrayList<SimpleString>(); + + byte low = (byte)(delim & 0xFF); // low byte + byte high = (byte)(delim >> 8 & 0xFF); // high byte + int lasPos = 0; for (int i = 0; i < data.length; i += 2) { - byte low = (byte)(delim & 0xFF); // low byte - byte high = (byte)(delim >> 8 & 0xFF); // high byte if (data[i] == low && data[i + 1] == high) { byte[] bytes = new byte[i - lasPos]; @@ -301,10 +302,11 @@ */ public boolean contains(final char c) { + final byte low = (byte)(c & 0xFF); // low byte + final byte high = (byte)(c >> 8 & 0xFF); // high byte + for (int i = 0; i < data.length; i += 2) { - byte low = (byte)(c & 0xFF); // low byte - byte high = (byte)(c >> 8 & 0xFF); // high byte if (data[i] == low && data[i + 1] == high) { return true; @@ -314,10 +316,9 @@ } /** - * concatanates a SimpleString and a String - * - * @param toAdd the String to concate with. - * @return the concatanated SimpleString + * Concatenates a SimpleString and a String + * @param toAdd the String to concatenate with. + * @return the concatenated SimpleString */ public SimpleString concat(final String toAdd) { @@ -325,10 +326,9 @@ } /** - * concatanates 2 SimpleString's - * - * @param toAdd the SimpleString to concate with. - * @return the concatanated SimpleString + * Concatenates 2 SimpleString's + * @param toAdd the SimpleString to concatenate with. + * @return the concatenated SimpleString */ public SimpleString concat(final SimpleString toAdd) { @@ -339,10 +339,9 @@ } /** - * concatanates a SimpleString and a char - * + * Concatenates a SimpleString and a char * @param c the char to concate with. - * @return the concatanated SimpleString + * @return the concatenated SimpleString */ public SimpleString concat(final char c) { @@ -390,7 +389,7 @@ } /** - * + * * @param srcBegin * @param srcEnd * @param dst Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -1146,8 +1146,15 @@ } } + HashMap<String, String> metaDataToSend; + + synchronized (metadata) + { + metaDataToSend = new HashMap<String, String>(metadata); + } + // Resetting the metadata after failover - for (Map.Entry<String, String> entries : metadata.entrySet()) + for (Map.Entry<String, String> entries : metaDataToSend.entrySet()) { sendPacketWithoutLock(new SessionAddMetaDataMessageV2(entries.getKey(), entries.getValue(), false)); } Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/RemotingService.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/RemotingService.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/RemotingService.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -16,7 +16,6 @@ import java.util.Set; import org.hornetq.api.core.Interceptor; -import org.hornetq.core.server.HornetQComponent; import org.hornetq.spi.core.protocol.RemotingConnection; /** @@ -25,7 +24,7 @@ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * @version <tt>$Revision$</tt> */ -public interface RemotingService extends HornetQComponent +public interface RemotingService { /** * Remove a connection from the connections held by the remoting service. @@ -41,7 +40,13 @@ void addInterceptor(Interceptor interceptor); boolean removeInterceptor(Interceptor interceptor); + + void stop(boolean criticalError) throws Exception; + + void start() throws Exception; + boolean isStarted(); + void freeze(); RemotingConnection getServerSideReplicatingConnection(); Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -263,7 +263,7 @@ } } - public void stop() throws Exception + public void stop(final boolean criticalError) throws Exception { if (!started) { @@ -275,7 +275,7 @@ return; } - failureCheckAndFlushThread.close(); + failureCheckAndFlushThread.close(criticalError); // We need to stop them accepting first so no new connections are accepted after we send the disconnect message for (Acceptor acceptor : acceptors) @@ -292,9 +292,13 @@ log.debug("Sending disconnect on live connections"); } + HashSet<ConnectionEntry> connectionEntries = new HashSet<ConnectionEntry>(); + + connectionEntries.addAll(connections.values()); + // Now we ensure that no connections will process any more packets after this method is complete // then send a disconnect packet - for (ConnectionEntry entry : connections.values()) + for (ConnectionEntry entry : connectionEntries) { RemotingConnection conn = entry.connection; @@ -321,12 +325,15 @@ } threadPool.shutdown(); - - boolean ok = threadPool.awaitTermination(10000, TimeUnit.MILLISECONDS); - - if (!ok) + + if (!criticalError) { - log.warn("Timed out waiting for remoting thread pool to terminate"); + boolean ok = threadPool.awaitTermination(10000, TimeUnit.MILLISECONDS); + + if (!ok) + { + log.warn("Timed out waiting for remoting thread pool to terminate"); + } } started = false; @@ -535,7 +542,7 @@ this.pauseInterval = pauseInterval; } - public void close() + public void close(final boolean criticalError) { closed = true; @@ -544,13 +551,16 @@ notify(); } - try + if (!criticalError) { - join(); + try + { + join(); + } + catch (InterruptedException ignore) + { + } } - catch (InterruptedException ignore) - { - } } @Override Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -493,7 +493,7 @@ } - remotingService.stop(); + remotingService.stop(criticalIOError); // We close all the exception in an attempt to let any pending IO to finish // to avoid scenarios where the send or ACK got to disk but the response didn't get to the client Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/QueueImpl.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -1222,6 +1222,7 @@ try { + boolean expired = false; while (iter.hasNext()) { MessageReference ref = iter.next(); @@ -1230,6 +1231,7 @@ if (ref.getMessage().isExpired()) { deliveringCount.incrementAndGet(); + expired = true; expire(ref); iter.remove(); refRemoved(ref); @@ -1240,6 +1242,11 @@ log.warn("Error expiring reference " + ref, e); } } + + if (expired && pageIterator != null && pageIterator.hasNext()) + { + scheduleDepage(); + } } finally { Modified: branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -1105,12 +1105,17 @@ public void send(final ServerMessage message, final boolean direct) throws Exception { long id = storageManager.generateUniqueID(); - + SimpleString address = message.getAddress(); message.setMessageID(id); message.encodeMessageIDToBuffer(); + if (defaultAddress == null && address != null) + { + defaultAddress = address; + } + if (address == null) { if (message.isDurable()) @@ -1131,6 +1136,12 @@ log.trace("send(message=" + message + ", direct=" + direct + ") being called"); } + if (message.getAddress() == null) + { + // This could happen with some tests that are ignoring messages + throw new HornetQException(HornetQException.ILLEGAL_STATE, "You don't have an address at the Server's Session"); + } + if (message.getAddress().equals(managementAddress)) { // It's a management message @@ -1141,11 +1152,6 @@ { doSend(message, direct); } - - if (defaultAddress == null) - { - defaultAddress = address; - } } public void sendContinuations(final int packetSize, Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ConsumerWindowSizeTest.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -303,7 +303,7 @@ consumers.add(consumer); session1.start(); sessions.add(session1); - consumer.receive(10); + assertNull(consumer.receive(10)); } @@ -321,7 +321,7 @@ senderSession.start(); ClientConsumer consumer = consumers.get(2); - ClientMessage received = consumer.receiveImmediate(); + ClientMessage received = consumer.receive(5000); assertNotNull(received); for (ClientSession tmpSess : sessions) Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/ExpiryLargeMessageTest.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -23,7 +23,9 @@ import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.api.core.client.ServerLocator; +import org.hornetq.core.logging.Logger; import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.Queue; import org.hornetq.core.settings.impl.AddressFullMessagePolicy; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.tests.util.ServiceTestBase; @@ -38,6 +40,8 @@ public class ExpiryLargeMessageTest extends ServiceTestBase { + private static final Logger log = Logger.getLogger(ExpiryLargeMessageTest.class); + // Constants ----------------------------------------------------- final SimpleString EXPIRY = new SimpleString("my-expiry"); @@ -129,25 +133,32 @@ producer.send(message); } + + session.close(); server.stop(); server.start(); + + Queue queueExpiry = server.locateQueue(EXPIRY); + Queue myQueue = server.locateQueue(MY_QUEUE); sf = locator.createSessionFactory(); - - session = sf.createSession(true, true, 0); - + Thread.sleep(1500); - // just to try expiring - ClientConsumer cons = session.createConsumer(MY_QUEUE); - assertNull(cons.receive(1000)); + long timeout = System.currentTimeMillis() + 5000; + while (timeout > System.currentTimeMillis() && queueExpiry.getMessageCount() != numberOfMessages) + { + // What the Expiry Scan would be doing + myQueue.expireReferences(); + Thread.sleep(50); + } + + assertEquals(50, queueExpiry.getMessageCount()); - session.close(); - session = sf.createSession(false, false); - cons = session.createConsumer(EXPIRY); + ClientConsumer cons = session.createConsumer(EXPIRY); session.start(); // Consume half of the messages to make sure all the messages are paging (on the second try) @@ -167,7 +178,7 @@ cons = session.createConsumer(EXPIRY); session.start(); - System.out.println("Trying " + rep); + log.info("Trying " + rep); for (int i = 0; i < numberOfMessages / 2; i++) { ClientMessage message = cons.receive(5000); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -4350,26 +4350,7 @@ session.start(); - // ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS.concat("=1")); - // - // for (int i = 0; i < numberOfMessages; i++) - // { - // message = consumer.receive(500000); - // assertNotNull(message); - // message.acknowledge(); - // - // // assertEquals(msg, message.getIntProperty("propTest").intValue()); - // - // System.out.println("i = " + i + " msg = " + message.getIntProperty("propTest")); - // } - // - // session.commit(); - - // consumer.close(); - session.deleteQueue(PagingTest.ADDRESS.concat("=1")); - // server.stop(); - // server.start(); sf = locator.createSessionFactory(); @@ -4385,8 +4366,6 @@ assertNotNull(message); message.acknowledge(); - // assertEquals(msg, message.getIntProperty("propTest").intValue()); - System.out.println("i = " + i + " msg = " + message.getIntProperty("propTest")); } @@ -4395,8 +4374,17 @@ assertNull(consumer.receiveImmediate()); consumer.close(); + + long timeout = System.currentTimeMillis() + 10000; + + PagingStore store = server.getPagingManager().getPageStore(ADDRESS); // It's async, so need to wait a bit for it happening + while (timeout > System.currentTimeMillis() && store.isPaging()) + { + Thread.sleep(100); + } + assertFalse(server.getPagingManager().getPageStore(ADDRESS).isPaging()); server.stop(); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/clientcrash/ClientTestBase.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/clientcrash/ClientTestBase.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/clientcrash/ClientTestBase.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -68,11 +68,21 @@ protected void assertActiveConnections(final int expectedActiveConnections) throws Exception { + long timeout = System.currentTimeMillis() + 5000; + while (timeout > System.currentTimeMillis() && server.getHornetQServerControl().getConnectionCount() != expectedActiveConnections) + { + Thread.sleep(100); + } Assert.assertEquals(expectedActiveConnections, server.getHornetQServerControl().getConnectionCount()); } protected void assertActiveSession(final int expectedActiveSession) throws Exception { + long timeout = System.currentTimeMillis() + 5000; + while (timeout > System.currentTimeMillis() && server.getSessions().size() != expectedActiveSession) + { + Thread.sleep(100); + } Assert.assertEquals(expectedActiveSession, server.getSessions().size()); } Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -291,6 +291,12 @@ ClientMessage message = session.createMessage(false); message.putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, System.currentTimeMillis() + delay); producer.send(message); + + long timeout = System.currentTimeMillis() + 5000; + while (timeout > System.currentTimeMillis() && queueControl.getScheduledCount() != 1) + { + Thread.sleep(100); + } Assert.assertEquals(1, queueControl.getScheduledCount()); ManagementTestBase.consumeMessages(0, session, queue); Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/LinkedListTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/LinkedListTest.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/LinkedListTest.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -13,6 +13,8 @@ package org.hornetq.tests.unit.util; +import java.lang.ref.WeakReference; +import java.util.LinkedList; import java.util.List; import java.util.NoSuchElementException; import java.util.concurrent.atomic.AtomicInteger; @@ -90,28 +92,12 @@ if (i % 1000 == 0) { System.out.println("Checking on " + i); - - for (int gcLoop = 0 ; gcLoop < 5; gcLoop++) - { - forceGC(); - if (count.get() == 1000) - { - break; - } - else - { - System.out.println("Trying a GC again"); - } - } - - assertEquals(1000, count.get()); + assertCount(1000, count); } } - forceGC(); + assertCount(1000, count); - assertEquals(1000, count.get()); - int removed = 0; while (iter.hasNext()) { @@ -119,11 +105,44 @@ iter.next(); iter.remove(); } + + assertCount(0, count); - forceGC(); + } - assertEquals(0, count.get()); + /** + * @param count + */ + private void assertCount(final int expected, final AtomicInteger count) + { + long timeout = System.currentTimeMillis() + 15000; + + int seqCount = 0; + while (timeout > System.currentTimeMillis() && count.get() != expected) + { + seqCount ++; + if (seqCount > 5) + { + LinkedList<String> toOME = new LinkedList<String>(); + int someCount = 0; + try + { + WeakReference<Object> ref = new WeakReference<Object>(new Object()); + while (ref.get() != null) + { + toOME.add("sdlfkjshadlfkjhas dlfkjhas dlfkjhads lkjfhads lfkjhads flkjashdf " + someCount++); + } + } + catch (Throwable expectedThrowable) + { + } + + toOME.clear(); + } + forceGC(); + } + assertEquals(expected, count.get()); } public void testAddTail() Modified: branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/SimpleStringTest.java =================================================================== --- branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/SimpleStringTest.java 2012-01-11 15:24:12 UTC (rev 12007) +++ branches/Branch_2_2_AS7/tests/src/org/hornetq/tests/unit/util/SimpleStringTest.java 2012-01-11 21:25:18 UTC (rev 12008) @@ -23,14 +23,56 @@ import org.hornetq.utils.DataConstants; /** - * + * * A SimpleStringTest - * + * * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> * */ public class SimpleStringTest extends UnitTestCase { + /** + * Converting back and forth between char and byte requires care as char is unsigned. + * @see SimpleString#getChars(int, int, char[], int) + * @see SimpleString#charAt(int) + * @see SimpleString#split(char) + */ + public void testGetChar() + { + SimpleString p1 = new SimpleString("foo"); + SimpleString p2 = new SimpleString("bar"); + for (int i = 0; i < 1 << 16; i++) + { + String msg = "expecting " + i; + char c = (char)i; + SimpleString s = new SimpleString(String.valueOf(c)); + + char[] c1 = new char[1]; + s.getChars(0, 1, c1, 0); + assertEquals(msg, c, c1[0]); + assertEquals(msg, c, s.charAt(0)); + SimpleString s2 = s.concat(c); + assertEquals(msg, c, s2.charAt(1)); + + // test splitting with chars + SimpleString sSplit = new SimpleString("foo" + String.valueOf(c) + "bar"); + SimpleString[] chunks = sSplit.split(c); + SimpleString[] split1 = p1.split(c); + SimpleString[] split2 = p2.split(c); + assertEquals(split1.length + split2.length, chunks.length); + int j = 0; + + for (SimpleString iS : split1) + { + assertEquals(iS.toString(), iS, chunks[j++]); + } + for (SimpleString iS : split2) + { + assertEquals(iS.toString(), iS, chunks[j++]); + } + } + } + public void testString() throws Exception { final String str = "hello123ABC__524`16254`6125!%^$!%$!%$!%$!%!$%!$$!\uA324";