[hornetq-commits] JBoss hornetq SVN: r11640 - in branches/Branch_2_2_EAP: src/main/org/hornetq/core/journal/impl and 5 other directories.

Author: clebert.suconic(a)jboss.com Date: 2011-11-02 12:57:20 -0400 (Wed, 02 Nov 2011) New Revision: 11640 Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFile.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/UnitTestCase.java Log: Fixing the testsuite on windows Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFile.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFile.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFile.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -93,7 +93,7 @@ void renameTo(String newFileName) throws Exception; - SequentialFile copy(); + SequentialFile cloneFile(); void copyTo(SequentialFile newFileName) throws Exception; Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -86,7 +86,7 @@ return pos; } - public SequentialFile copy() + public SequentialFile cloneFile() { return new AIOSequentialFile(factory, -1, Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -109,7 +109,10 @@ { log.debug("Copying " + this + " as " + newFileName); newFileName.open(); - this.open(); + if (!isOpen()) + { + this.open(); + } ByteBuffer buffer = ByteBuffer.allocate(10 * 1024); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -252,7 +252,7 @@ return "NIOSequentialFile " + getFile(); } - public SequentialFile copy() + public SequentialFile cloneFile() { return new NIOSequentialFile(factory, getFile(), maxIO, writerExecutor); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/impl/PageImpl.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -212,7 +212,10 @@ public void open() throws Exception { - file.open(); + if (!file.isOpen()) + { + file.open(); + } size.set((int)file.size()); file.position(0); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -383,6 +383,10 @@ log.warn("Error on copying large message this for DLA or Expiry", e); return null; } + finally + { + releaseResources(); + } } } @@ -426,7 +430,7 @@ file = storageManager.createFileForLargeMessage(getMessageID(), durable); - file.open(); + openFile(); bodySize = file.size(); } @@ -437,6 +441,27 @@ throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e); } } + + protected void openFile() throws Exception + { + if (file == null) + { + validateFile(); + } + else + if (!file.isOpen()) + { + file.open(); + } + } + + protected void closeFile() throws Exception + { + if (file != null && file.isOpen()) + { + file.close(); + } + } /* (non-Javadoc) * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) @@ -454,9 +479,9 @@ file = storageManager.createFileForLargeMessage(message.getMessageID(), durable); try { - file.open(); + openFile(); bodySize = file.size(); - file.close(); + closeFile(); } catch (Exception e) { @@ -474,7 +499,11 @@ { try { - cFile = file.copy(); + if (cFile != null && cFile.isOpen()) + { + cFile.close(); + } + cFile = file.cloneFile(); cFile.open(); } catch (Exception e) Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -121,6 +121,8 @@ protected void tearDown() throws Exception { locator.close(); + + locator = null; super.tearDown(); } @@ -263,7 +265,7 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); - + waitForServer(server); queue = server.locateQueue(ADDRESS); @@ -647,8 +649,7 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); - - + locator = createInVMNonHALocator(); locator.setBlockOnNonDurableSend(true); locator.setBlockOnDurableSend(true); @@ -662,7 +663,7 @@ session = sf.createSession(false, false, false); producer = session.createProducer(PagingTest.ADDRESS); - + for (int i = 0; i < numberOfMessages; i++) { message = session.createMessage(true); @@ -679,9 +680,9 @@ session.commit(); } } - + session.commit(); - + server.stop(); server = createServer(true, @@ -696,7 +697,7 @@ queue = server.locateQueue(ADDRESS); - // assertEquals(numberOfMessages, queue.getMessageCount()); + // assertEquals(numberOfMessages, queue.getMessageCount()); xids = new LinkedList<Xid>(); @@ -728,7 +729,6 @@ sessionConsumer.close(); - } finally { @@ -1041,50 +1041,56 @@ server.start(); - ServerLocator locator = createInVMNonHALocator(); + try + { - locator.setBlockOnNonDurableSend(true); - locator.setBlockOnDurableSend(true); - locator.setBlockOnAcknowledge(true); + ServerLocator locator = createInVMNonHALocator(); - ClientSessionFactory csf = locator.createSessionFactory(); + locator.setBlockOnNonDurableSend(true); + locator.setBlockOnDurableSend(true); + locator.setBlockOnAcknowledge(true); - ClientSession session = csf.createSession(); + ClientSessionFactory csf = locator.createSessionFactory(); - session.start(); + ClientSession session = csf.createSession(); - for (int i = 1; i <= 2; i++) - { - ClientConsumer cons = session.createConsumer("q" + i); + session.start(); - for (int j = 3; j < 6; j++) + for (int i = 1; i <= 2; i++) { - ClientMessage msg = cons.receive(5000); + ClientConsumer cons = session.createConsumer("q" + i); - assertNotNull(msg); + for (int j = 3; j < 6; j++) + { + ClientMessage msg = cons.receive(5000); - assertEquals("str-" + j, msg.getStringProperty("id")); + assertNotNull(msg); - msg.acknowledge(); + assertEquals("str-" + j, msg.getStringProperty("id")); + + msg.acknowledge(); + } + + session.commit(); + assertNull(cons.receive(500)); + } - session.commit(); - assertNull(cons.receive(500)); + session.close(); - } + long timeout = System.currentTimeMillis() + 5000; - session.close(); + while (System.currentTimeMillis() < timeout && server.getPagingManager().getPageStore(ADDRESS).isPaging()) + { + Thread.sleep(100); + } - long timeout = System.currentTimeMillis() + 5000; - - while (System.currentTimeMillis() < timeout && server.getPagingManager().getPageStore(ADDRESS).isPaging()) + locator.close(); + } + finally { - Thread.sleep(100); + server.stop(); } - - locator.close(); - - server.stop(); } public void testTwoQueuesOneNoRouting() throws Exception @@ -1277,27 +1283,28 @@ { bb.put(getSamplebyte(j)); } - + final AtomicBoolean running = new AtomicBoolean(true); - + class TCount extends Thread { Queue queue; - + TCount(Queue queue) { this.queue = queue; } + public void run() { try { while (running.get()) { - // log.info("Message count = " + queue.getMessageCount() + " on queue " + queue.getName()); + // log.info("Message count = " + queue.getMessageCount() + " on queue " + queue.getName()); queue.getMessagesAdded(); queue.getMessageCount(); - //log.info("Message added = " + queue.getMessagesAdded() + " on queue " + queue.getName()); + // log.info("Message added = " + queue.getMessagesAdded() + " on queue " + queue.getName()); Thread.sleep(10); } } @@ -1307,10 +1314,9 @@ } } }; - + TCount tcount1 = null; TCount tcount2 = null; - try { @@ -1337,8 +1343,7 @@ session.createQueue(PagingTest.ADDRESS.toString(), PagingTest.ADDRESS + "-2", null, true); } - - + ClientProducer producer = session.createProducer(PagingTest.ADDRESS); ClientMessage message = null; @@ -1377,21 +1382,21 @@ PagingTest.PAGE_MAX, new HashMap<String, AddressSettings>()); server.start(); - + Queue queue1 = server.locateQueue(PagingTest.ADDRESS.concat("-1")); - + Queue queue2 = server.locateQueue(PagingTest.ADDRESS.concat("-2")); - + assertNotNull(queue1); - + assertNotNull(queue2); - + assertNotSame(queue1, queue2); tcount1 = new TCount(queue1); - + tcount2 = new TCount(queue2); - + tcount1.start(); tcount2.start(); @@ -1500,19 +1505,19 @@ finally { running.set(false); - + if (tcount1 != null) { tcount1.interrupt(); tcount1.join(); } - + if (tcount2 != null) { tcount2.interrupt(); tcount2.join(); } - + try { server.stop(); @@ -2510,7 +2515,7 @@ producerThread.start(); - assertTrue(ready.await(10, TimeUnit.SECONDS)); + assertTrue(ready.await(100, TimeUnit.SECONDS)); ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -609,7 +609,7 @@ { } - public SequentialFile copy() + public SequentialFile cloneFile() { return null; // To change body of implemented methods use File | Settings | File Templates. } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/UnitTestCase.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-11-02 16:19:21 UTC (rev 11639) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/util/UnitTestCase.java 2011-11-02 16:57:20 UTC (rev 11640) @@ -1174,7 +1174,9 @@ File fileTmp = new File(directory, files[j]); if (!deleteDirectory(fileTmp)) { + // This is because of Windows is dumb on releasing files log.warn("Couldn't delete " + fileTmp); + forceGC(); return false; } }