You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2016/03/04 23:42:50 UTC
[06/58] [abbrv] activemq-artemis git commit: open wire changes
equivalent to ab16f7098fb52d2b4c40627ed110e1776525f208
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/perf/TempKahaStoreQueueTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/perf/TempKahaStoreQueueTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/perf/TempKahaStoreQueueTest.java
deleted file mode 100644
index 2575afd..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/perf/TempKahaStoreQueueTest.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.store.kahadb.perf;
-
-import java.io.File;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.perf.SimpleQueueTest;
-import org.apache.activemq.store.kahadb.TempKahaDBStore;
-
-/**
- *
- */
-public class TempKahaStoreQueueTest extends SimpleQueueTest {
-
- @Override
- protected void configureBroker(BrokerService answer, String uri) throws Exception {
- File dataFileDir = new File("target/test-amq-data/perfTest/temp-amqdb");
- dataFileDir.mkdirs();
- answer.setDeleteAllMessagesOnStartup(true);
-
- TempKahaDBStore adaptor = new TempKahaDBStore();
- adaptor.setDirectory(dataFileDir);
-
- answer.setDataDirectoryFile(dataFileDir);
- answer.setPersistenceAdapter(adaptor);
- answer.addConnector(uri);
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/KahaDBFilePendingMessageCursorTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/KahaDBFilePendingMessageCursorTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/KahaDBFilePendingMessageCursorTest.java
deleted file mode 100644
index 6626603..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/KahaDBFilePendingMessageCursorTest.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.store.kahadb.plist;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.region.*;
-import org.apache.activemq.broker.region.cursors.FilePendingMessageCursor;
-import org.apache.activemq.broker.region.cursors.FilePendingMessageCursorTestSupport;
-import org.apache.activemq.command.ActiveMQMessage;
-import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.activemq.command.MessageId;
-import org.apache.activemq.store.kahadb.disk.page.PageFile;
-import org.apache.activemq.usage.SystemUsage;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-
-/**
- * @author <a href="http://hiramchirino.com">Hiram Chirino</a>
- */
-public class KahaDBFilePendingMessageCursorTest extends FilePendingMessageCursorTestSupport {
-
- @Test
- public void testAddRemoveAddIndexSize() throws Exception {
- brokerService = new BrokerService();
- brokerService.setUseJmx(false);
- SystemUsage usage = brokerService.getSystemUsage();
- usage.getMemoryUsage().setLimit(1024 * 150);
- String body = new String(new byte[1024]);
- Destination destination = new Queue(brokerService, new ActiveMQQueue("Q"), null, new DestinationStatistics(), null);
-
- underTest = new FilePendingMessageCursor(brokerService.getBroker(), "test", false);
- underTest.setSystemUsage(usage);
-
- LOG.info("start");
- final PageFile pageFile = ((PListImpl) underTest.getDiskList()).getPageFile();
- LOG.info("page count: " + pageFile.getPageCount());
- LOG.info("free count: " + pageFile.getFreePageCount());
- LOG.info("content size: " + pageFile.getPageContentSize());
-
- final long initialPageCount = pageFile.getPageCount();
-
- final int numMessages = 1000;
-
- for (int j = 0; j < 10; j++) {
- // ensure free pages are reused
- for (int i = 0; i < numMessages; i++) {
- ActiveMQMessage mqMessage = new ActiveMQMessage();
- mqMessage.setStringProperty("body", body);
- mqMessage.setMessageId(new MessageId("1:2:3:" + i));
- mqMessage.setMemoryUsage(usage.getMemoryUsage());
- mqMessage.setRegionDestination(destination);
- underTest.addMessageLast(new IndirectMessageReference(mqMessage));
- }
- assertFalse("cursor is not full " + usage.getTempUsage(), underTest.isFull());
-
- underTest.reset();
- long receivedCount = 0;
- while (underTest.hasNext()) {
- MessageReference ref = underTest.next();
- underTest.remove();
- ref.decrementReferenceCount();
- assertEquals("id is correct", receivedCount++, ref.getMessageId().getProducerSequenceId());
- }
- assertEquals("got all messages back", receivedCount, numMessages);
- LOG.info("page count: " + pageFile.getPageCount());
- LOG.info("free count: " + pageFile.getFreePageCount());
- LOG.info("content size: " + pageFile.getPageContentSize());
- }
-
- assertEquals("expected page usage", initialPageCount, pageFile.getPageCount() - pageFile.getFreePageCount());
-
- LOG.info("Destroy");
- underTest.destroy();
- LOG.info("page count: " + pageFile.getPageCount());
- LOG.info("free count: " + pageFile.getFreePageCount());
- LOG.info("content size: " + pageFile.getPageContentSize());
- assertEquals("expected page usage", initialPageCount - 1, pageFile.getPageCount() - pageFile.getFreePageCount());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/PListTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/PListTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/PListTest.java
deleted file mode 100644
index 73adb52..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/plist/PListTest.java
+++ /dev/null
@@ -1,669 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.store.kahadb.plist;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.activemq.store.PList;
-import org.apache.activemq.store.PListEntry;
-import org.apache.activemq.util.ByteSequence;
-import org.apache.activemq.util.IOHelper;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class PListTest {
-
- static final Logger LOG = LoggerFactory.getLogger(PListTest.class);
- private PListStoreImpl store;
- private PListImpl plist;
- final ByteSequence payload = new ByteSequence(new byte[400]);
- final String idSeed = new String("Seed" + Arrays.toString(new byte[1024]));
- final Vector<Throwable> exceptions = new Vector<>();
- ExecutorService executor;
-
- private PListEntry getFirst(PList plist) throws IOException {
- PList.PListIterator iterator = plist.iterator();
- try {
- if (iterator.hasNext()) {
- return iterator.next();
- }
- else {
- return null;
- }
- }
- finally {
- iterator.release();
- }
- }
-
- @Test
- public void testAddLast() throws Exception {
- final int COUNT = 1000;
- Map<String, ByteSequence> map = new LinkedHashMap<>();
- for (int i = 0; i < COUNT; i++) {
- String test = new String("test" + i);
- ByteSequence bs = new ByteSequence(test.getBytes());
- map.put(test, bs);
- plist.addLast(test, bs);
- }
- assertEquals(plist.size(), COUNT);
- int count = 0;
- for (ByteSequence bs : map.values()) {
- String origStr = new String(bs.getData(), bs.getOffset(), bs.getLength());
- PListEntry entry = plist.get(count);
- String plistString = new String(entry.getByteSequence().getData(), entry.getByteSequence().getOffset(), entry.getByteSequence().getLength());
- assertEquals(origStr, plistString);
- count++;
- }
- }
-
- @Test
- public void testAddFirst() throws Exception {
- final int COUNT = 1000;
- Map<String, ByteSequence> map = new LinkedHashMap<>();
- for (int i = 0; i < COUNT; i++) {
- String test = new String("test" + i);
- ByteSequence bs = new ByteSequence(test.getBytes());
- map.put(test, bs);
- plist.addFirst(test, bs);
- }
- assertEquals(plist.size(), COUNT);
- long count = plist.size() - 1;
- for (ByteSequence bs : map.values()) {
- String origStr = new String(bs.getData(), bs.getOffset(), bs.getLength());
- PListEntry entry = plist.get(count);
- String plistString = new String(entry.getByteSequence().getData(), entry.getByteSequence().getOffset(), entry.getByteSequence().getLength());
- assertEquals(origStr, plistString);
- count--;
- }
- }
-
- @Test
- public void testRemove() throws IOException {
- doTestRemove(2000);
- }
-
- protected void doTestRemove(final int COUNT) throws IOException {
- Map<String, ByteSequence> map = new LinkedHashMap<>();
- for (int i = 0; i < COUNT; i++) {
- String test = new String("test" + i);
- ByteSequence bs = new ByteSequence(test.getBytes());
- map.put(test, bs);
- plist.addLast(test, bs);
- }
- assertEquals(plist.size(), COUNT);
- PListEntry entry = plist.getFirst();
- while (entry != null) {
- plist.remove(entry.getId());
- entry = plist.getFirst();
- }
- assertEquals(0, plist.size());
- }
-
- @Test
- public void testDestroy() throws Exception {
- doTestRemove(1);
- plist.destroy();
- assertEquals(0, plist.size());
- }
-
- @Test
- public void testDestroyNonEmpty() throws Exception {
- final int COUNT = 1000;
- Map<String, ByteSequence> map = new LinkedHashMap<>();
- for (int i = 0; i < COUNT; i++) {
- String test = new String("test" + i);
- ByteSequence bs = new ByteSequence(test.getBytes());
- map.put(test, bs);
- plist.addLast(test, bs);
- }
- plist.destroy();
- assertEquals(0, plist.size());
- }
-
- @Test
- public void testRemoveSecond() throws Exception {
- plist.addLast("First", new ByteSequence("A".getBytes()));
- plist.addLast("Second", new ByteSequence("B".getBytes()));
-
- assertTrue(plist.remove("Second"));
- assertTrue(plist.remove("First"));
- assertFalse(plist.remove("doesNotExist"));
- }
-
- @Test
- public void testRemoveSingleEntry() throws Exception {
- plist.addLast("First", new ByteSequence("A".getBytes()));
-
- Iterator<PListEntry> iterator = plist.iterator();
- while (iterator.hasNext()) {
- iterator.next();
- iterator.remove();
- }
- }
-
- @Test
- public void testRemoveSecondPosition() throws Exception {
- plist.addLast("First", new ByteSequence("A".getBytes()));
- plist.addLast("Second", new ByteSequence("B".getBytes()));
-
- assertTrue(plist.remove(1));
- assertTrue(plist.remove(0));
- assertFalse(plist.remove(0));
- }
-
- @Test
- public void testConcurrentAddRemove() throws Exception {
- File directory = store.getDirectory();
- store.stop();
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- store = new PListStoreImpl();
- store.setCleanupInterval(400);
- store.setDirectory(directory);
- store.setJournalMaxFileLength(1024 * 5);
- store.setLazyInit(false);
- store.start();
-
- final ByteSequence payload = new ByteSequence(new byte[1024 * 2]);
-
- final Vector<Throwable> exceptions = new Vector<>();
- final int iterations = 1000;
- final int numLists = 10;
-
- final PList[] lists = new PList[numLists];
- String threadName = Thread.currentThread().getName();
- for (int i = 0; i < numLists; i++) {
- Thread.currentThread().setName("C:" + String.valueOf(i));
- lists[i] = store.getPList(String.valueOf(i));
- }
- Thread.currentThread().setName(threadName);
-
- executor = Executors.newFixedThreadPool(100);
- class A implements Runnable {
-
- @Override
- public void run() {
- final String threadName = Thread.currentThread().getName();
- try {
- for (int i = 0; i < iterations; i++) {
- PList candidate = lists[i % numLists];
- Thread.currentThread().setName("ALRF:" + candidate.getName());
- synchronized (plistLocks(candidate)) {
- Object locator = candidate.addLast(String.valueOf(i), payload);
- getFirst(candidate);
- assertTrue(candidate.remove(locator));
- }
- }
- }
- catch (Exception error) {
- LOG.error("Unexpcted ex", error);
- error.printStackTrace();
- exceptions.add(error);
- }
- finally {
- Thread.currentThread().setName(threadName);
- }
- }
- }
-
- class B implements Runnable {
-
- @Override
- public void run() {
- final String threadName = Thread.currentThread().getName();
- try {
- for (int i = 0; i < iterations; i++) {
- PList candidate = lists[i % numLists];
- Thread.currentThread().setName("ALRF:" + candidate.getName());
- synchronized (plistLocks(candidate)) {
- Object locator = candidate.addLast(String.valueOf(i), payload);
- getFirst(candidate);
- assertTrue(candidate.remove(locator));
- }
- }
- }
- catch (Exception error) {
- error.printStackTrace();
- exceptions.add(error);
- }
- finally {
- Thread.currentThread().setName(threadName);
- }
- }
- }
-
- executor.execute(new A());
- executor.execute(new A());
- executor.execute(new A());
- executor.execute(new B());
- executor.execute(new B());
- executor.execute(new B());
-
- executor.shutdown();
- boolean finishedInTime = executor.awaitTermination(30, TimeUnit.SECONDS);
-
- assertTrue("no exceptions", exceptions.isEmpty());
- assertTrue("finished ok", finishedInTime);
- }
-
- @Test
- public void testConcurrentAddLast() throws Exception {
- File directory = store.getDirectory();
- store.stop();
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- store = new PListStoreImpl();
- store.setDirectory(directory);
- store.start();
-
- final int numThreads = 20;
- final int iterations = 1000;
- executor = Executors.newFixedThreadPool(100);
- for (int i = 0; i < numThreads; i++) {
- new Job(i, PListTest.TaskType.ADD, iterations).run();
- }
-
- for (int i = 0; i < numThreads; i++) {
- executor.execute(new Job(i, PListTest.TaskType.ITERATE, iterations));
- }
-
- for (int i = 0; i < 100; i++) {
- executor.execute(new Job(i + 20, PListTest.TaskType.ADD, 100));
- }
-
- executor.shutdown();
- boolean finishedInTime = executor.awaitTermination(60 * 5, TimeUnit.SECONDS);
- assertTrue("finished ok", finishedInTime);
- }
-
- @Test
- public void testOverFlow() throws Exception {
- File directory = store.getDirectory();
- store.stop();
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- store = new PListStoreImpl();
- store.setDirectory(directory);
- store.start();
-
- for (int i = 0; i < 2000; i++) {
- new Job(i, PListTest.TaskType.ADD, 5).run();
-
- }
- LOG.info("After Load index file: " + store.pageFile.getFile().length());
- LOG.info("After remove index file: " + store.pageFile.getFile().length());
- }
-
- @Test
- public void testConcurrentAddRemoveWithPreload() throws Exception {
- File directory = store.getDirectory();
- store.stop();
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- store = new PListStoreImpl();
- store.setDirectory(directory);
- store.setJournalMaxFileLength(1024 * 5);
- store.setCleanupInterval(5000);
- store.setIndexWriteBatchSize(500);
- store.start();
-
- final int iterations = 500;
- final int numLists = 10;
-
- // prime the store
-
- // create/delete
- LOG.info("create");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.CREATE, iterations).run();
- }
-
- LOG.info("delete");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.DELETE, iterations).run();
- }
-
- LOG.info("fill");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.ADD, iterations).run();
- }
- LOG.info("remove");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.REMOVE, iterations).run();
- }
-
- LOG.info("check empty");
- for (int i = 0; i < numLists; i++) {
- assertEquals("empty " + i, 0, store.getPList("List-" + i).size());
- }
-
- LOG.info("delete again");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.DELETE, iterations).run();
- }
-
- LOG.info("fill again");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.ADD, iterations).run();
- }
-
- LOG.info("parallel add and remove");
- executor = Executors.newFixedThreadPool(numLists * 2);
- for (int i = 0; i < numLists * 2; i++) {
- executor.execute(new Job(i, i >= numLists ? PListTest.TaskType.ADD : PListTest.TaskType.REMOVE, iterations));
- }
-
- executor.shutdown();
- LOG.info("wait for parallel work to complete");
- boolean finishedInTime = executor.awaitTermination(60 * 5, TimeUnit.SECONDS);
- assertTrue("no exceptions", exceptions.isEmpty());
- assertTrue("finished ok", finishedInTime);
- }
-
- // for non determinant issues, increasing this may help diagnose
- final int numRepeats = 1;
-
- @Test
- public void testRepeatStressWithCache() throws Exception {
- for (int i = 0; i < numRepeats; i++) {
- do_testConcurrentAddIterateRemove(true);
- }
- }
-
- @Test
- public void testRepeatStressWithOutCache() throws Exception {
- for (int i = 0; i < numRepeats; i++) {
- do_testConcurrentAddIterateRemove(false);
- }
- }
-
- public void do_testConcurrentAddIterateRemove(boolean enablePageCache) throws Exception {
- File directory = store.getDirectory();
- store.stop();
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- store = new PListStoreImpl();
- store.setIndexEnablePageCaching(enablePageCache);
- store.setIndexPageSize(2 * 1024);
- store.setDirectory(directory);
- store.start();
-
- final int iterations = 500;
- final int numLists = 10;
-
- LOG.info("create");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.CREATE, iterations).run();
- }
-
- LOG.info("fill");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.ADD, iterations).run();
- }
-
- LOG.info("parallel add and remove");
- executor = Executors.newFixedThreadPool(400);
- final int numProducer = 5;
- final int numConsumer = 10;
- for (int i = 0; i < numLists; i++) {
- for (int j = 0; j < numProducer; j++) {
- executor.execute(new Job(i, PListTest.TaskType.ADD, iterations * 2));
- }
- for (int k = 0; k < numConsumer; k++) {
- executor.execute(new Job(i, TaskType.ITERATE_REMOVE, iterations / 4));
- }
- }
-
- for (int i = numLists; i < numLists * 10; i++) {
- executor.execute(new Job(i, PListTest.TaskType.ADD, iterations));
- }
-
- executor.shutdown();
- LOG.info("wait for parallel work to complete");
- boolean shutdown = executor.awaitTermination(60 * 60, TimeUnit.SECONDS);
- assertTrue("no exceptions: " + exceptions, exceptions.isEmpty());
- assertTrue("test did not timeout ", shutdown);
- }
-
- @Test
- public void testConcurrentAddIterate() throws Exception {
- File directory = store.getDirectory();
- store.stop();
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- store = new PListStoreImpl();
- store.setIndexPageSize(2 * 1024);
- store.setJournalMaxFileLength(1024 * 1024);
- store.setDirectory(directory);
- store.setCleanupInterval(-1);
- store.setIndexEnablePageCaching(false);
- store.setIndexWriteBatchSize(100);
- store.start();
-
- final int iterations = 250;
- final int numLists = 10;
-
- LOG.info("create");
- for (int i = 0; i < numLists; i++) {
- new Job(i, PListTest.TaskType.CREATE, iterations).run();
- }
-
- LOG.info("parallel add and iterate");
- // We want a lot of adds occurring so that new free pages get created
- // along
- // with overlapping seeks from the iterators so that we are likely to
- // seek into
- // some bad area in the page file.
- executor = Executors.newFixedThreadPool(100);
- final int numProducer = 30;
- final int numConsumer = 10;
- for (int i = 0; i < numLists; i++) {
- for (int j = 0; j < numProducer; j++) {
- executor.execute(new Job(i, PListTest.TaskType.ADD, iterations));
- }
- for (int k = 0; k < numConsumer; k++) {
- executor.execute(new Job(i, TaskType.ITERATE, iterations * 2));
- }
- }
-
- executor.shutdown();
- LOG.info("wait for parallel work to complete");
- boolean shutdown = executor.awaitTermination(5 * 60, TimeUnit.SECONDS);
- assertTrue("no exceptions: " + exceptions, exceptions.isEmpty());
- assertTrue("test did not timeout ", shutdown);
- LOG.info("Num dataFiles:" + store.getJournal().getFiles().size());
- }
-
- enum TaskType {
- CREATE, DELETE, ADD, REMOVE, ITERATE, ITERATE_REMOVE
- }
-
- class Job implements Runnable {
-
- int id;
- TaskType task;
- int iterations;
-
- public Job(int id, TaskType t, int iterations) {
- this.id = id;
- this.task = t;
- this.iterations = iterations;
- }
-
- @Override
- public void run() {
- final String threadName = Thread.currentThread().getName();
- try {
- PListImpl plist = null;
- switch (task) {
- case CREATE:
- Thread.currentThread().setName("C:" + id);
- plist = store.getPList(String.valueOf(id));
- LOG.info("Job-" + id + ", CREATE");
- break;
- case DELETE:
- Thread.currentThread().setName("D:" + id);
- store.removePList(String.valueOf(id));
- break;
- case ADD:
- Thread.currentThread().setName("A:" + id);
- plist = store.getPList(String.valueOf(id));
-
- for (int j = 0; j < iterations; j++) {
- synchronized (plistLocks(plist)) {
- if (exceptions.isEmpty()) {
- plist.addLast("PL>" + id + idSeed + "-" + j, payload);
- }
- else {
- break;
- }
- }
- }
-
- if (exceptions.isEmpty()) {
- LOG.info("Job-" + id + ", Add, done: " + iterations);
- }
- break;
- case REMOVE:
- Thread.currentThread().setName("R:" + id);
- plist = store.getPList(String.valueOf(id));
- synchronized (plistLocks(plist)) {
-
- for (int j = iterations - 1; j >= 0; j--) {
- plist.remove("PL>" + id + idSeed + "-" + j);
- if (j > 0 && j % (iterations / 2) == 0) {
- LOG.info("Job-" + id + " Done remove: " + j);
- }
- }
- }
- break;
- case ITERATE:
- Thread.currentThread().setName("I:" + id);
- plist = store.getPList(String.valueOf(id));
- int iterateCount = 0;
- synchronized (plistLocks(plist)) {
- if (exceptions.isEmpty()) {
- Iterator<PListEntry> iterator = plist.iterator();
- while (iterator.hasNext() && exceptions.isEmpty()) {
- iterator.next();
- iterateCount++;
- }
-
- // LOG.info("Job-" + id + " Done iterate: it=" +
- // iterator + ", count:" + iterateCount +
- // ", size:" + plist.size());
- if (plist.size() != iterateCount) {
- System.err.println("Count Wrong: " + iterator);
- }
- assertEquals("iterate got all " + id + " iterator:" + iterator, plist.size(), iterateCount);
- }
- }
- break;
-
- case ITERATE_REMOVE:
- Thread.currentThread().setName("IRM:" + id);
- plist = store.getPList(String.valueOf(id));
-
- int removeCount = 0;
- synchronized (plistLocks(plist)) {
-
- Iterator<PListEntry> removeIterator = plist.iterator();
-
- while (removeIterator.hasNext()) {
- removeIterator.next();
- removeIterator.remove();
- if (removeCount++ > iterations) {
- break;
- }
- }
- }
- LOG.info("Job-" + id + " Done remove: " + removeCount);
- break;
-
- default:
- }
-
- }
- catch (Exception e) {
- LOG.warn("Job[" + id + "] caught exception: " + e.getMessage());
- e.printStackTrace();
- exceptions.add(e);
- if (executor != null) {
- executor.shutdownNow();
- }
- }
- finally {
- Thread.currentThread().setName(threadName);
- }
- }
- }
-
- final Map<PList, Object> locks = new HashMap<>();
-
- private Object plistLocks(PList plist) {
- Object lock = null;
- synchronized (locks) {
- if (locks.containsKey(plist)) {
- lock = locks.get(plist);
- }
- else {
- lock = new Object();
- locks.put(plist, lock);
- }
- }
- return lock;
- }
-
- @Before
- public void setUp() throws Exception {
- File directory = new File("target/test/PlistDB");
- IOHelper.mkdirs(directory);
- IOHelper.deleteChildren(directory);
- startStore(directory);
-
- }
-
- protected void startStore(File directory) throws Exception {
- store = new PListStoreImpl();
- store.setDirectory(directory);
- store.start();
- plist = store.getPList("main");
- }
-
- @After
- public void tearDown() throws Exception {
- if (executor != null) {
- executor.shutdownNow();
- }
- store.stop();
- exceptions.clear();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/shared.xml
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/shared.xml b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/shared.xml
deleted file mode 100644
index 5042df8..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/shared.xml
+++ /dev/null
@@ -1,59 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- Licensed to the Apache Software Foundation (ASF) under one or more
- contributor license agreements. See the NOTICE file distributed with
- this work for additional information regarding copyright ownership.
- The ASF licenses this file to You under the Apache License, Version 2.0
- (the "License"); you may not use this file except in compliance with
- the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
--->
-<beans
- xmlns="http://www.springframework.org/schema/beans"
- xmlns:amq="http://activemq.apache.org/schema/core"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
- http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
-
- <!-- normal ActiveMQ XML config which is less verbose & can be validated -->
- <amq:broker brokerName="brokerConfigTest" populateJMSXUserID="false"
- useLoggingForShutdownErrors="true" useJmx="true"
- persistent="true" vmConnectorURI="vm://javacoola"
- useShutdownHook="false" deleteAllMessagesOnStartup="true">
-
- <amq:persistenceAdapter>
- <amq:kahaDB directory = "target/activemq-data">
- <amq:locker>
- <amq:shared-file-locker lockAcquireSleepInterval="5000"/>
- </amq:locker>
- </amq:kahaDB>
- </amq:persistenceAdapter>
-
- <amq:systemUsage>
- <amq:systemUsage>
- <amq:memoryUsage>
- <amq:memoryUsage limit="10 mb" percentUsageMinDelta="20"/>
- </amq:memoryUsage>
- <amq:storeUsage>
- <amq:storeUsage limit="1 gb" name="foo"/>
- </amq:storeUsage>
- <amq:tempUsage>
- <amq:tempUsage limit="100 mb"/>
- </amq:tempUsage>
- </amq:systemUsage>
- </amq:systemUsage>
-
- <amq:transportConnectors>
- <amq:transportConnector uri="tcp://localhost:61635"/>
- </amq:transportConnectors>
-
- </amq:broker>
-
-</beans>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBNegativeQueueTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBNegativeQueueTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBNegativeQueueTest.java
deleted file mode 100644
index 3c28b3d..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBNegativeQueueTest.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.store.leveldb;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.region.cursors.NegativeQueueTest;
-import org.apache.activemq.leveldb.LevelDBStore;
-import org.apache.activemq.util.IOHelper;
-
-import java.io.File;
-
-public class LevelDBNegativeQueueTest extends NegativeQueueTest {
-
- @Override
- protected void configureBroker(BrokerService answer) throws Exception {
- super.configureBroker(answer);
- LevelDBStore levelDBStore = new LevelDBStore();
- File directory = new File("target/activemq-data/leveldb");
- IOHelper.deleteChildren(directory);
- levelDBStore.setDirectory(directory);
- levelDBStore.deleteAllMessages();
- answer.setPersistenceAdapter(levelDBStore);
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBStoreBrokerTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBStoreBrokerTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBStoreBrokerTest.java
deleted file mode 100644
index 99583d5..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/leveldb/LevelDBStoreBrokerTest.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.store.leveldb;
-
-import java.io.File;
-
-import junit.framework.Test;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.BrokerTest;
-import org.apache.activemq.store.kahadb.KahaDBStore;
-import org.apache.activemq.util.IOHelper;
-import org.apache.activemq.leveldb.LevelDBStore;
-
-/**
- * Once the wire format is completed we can test against real persistence storage.
- */
-public class LevelDBStoreBrokerTest extends BrokerTest {
-
- @Override
- protected void setUp() throws Exception {
- this.setAutoFail(true);
- super.setUp();
- }
-
- @Override
- protected BrokerService createBroker() throws Exception {
- BrokerService broker = new BrokerService();
- LevelDBStore levelDBStore = new LevelDBStore();
- File directory = new File("target/activemq-data/leveldb");
- IOHelper.deleteChildren(directory);
- levelDBStore.setDirectory(directory);
- levelDBStore.deleteAllMessages();
- broker.setPersistenceAdapter(levelDBStore);
- return broker;
- }
-
- protected BrokerService createRestartedBroker() throws Exception {
- BrokerService broker = new BrokerService();
- KahaDBStore kaha = new KahaDBStore();
- kaha.setDirectory(new File("target/activemq-data/leveldb"));
- broker.setPersistenceAdapter(kaha);
- return broker;
- }
-
- public static Test suite() {
- return suite(LevelDBStoreBrokerTest.class);
- }
-
- public static void main(String[] args) {
- junit.textui.TestRunner.run(suite());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/db-1.log
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/db-1.log b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/db-1.log
deleted file mode 100644
index e69de29..0000000
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/scheduleDB.data
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/scheduleDB.data b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/scheduleDB.data
deleted file mode 100644
index e69de29..0000000
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/scheduleDB.redo
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/scheduleDB.redo b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/schedulerDB/legacy/scheduleDB.redo
deleted file mode 100644
index e69de29..0000000
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java
deleted file mode 100644
index 38f0213..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/streams/JMSInputStreamTest.java
+++ /dev/null
@@ -1,286 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.streams;
-
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-
-import junit.framework.Test;
-
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQInputStream;
-import org.apache.activemq.ActiveMQOutputStream;
-import org.apache.activemq.JmsTestSupport;
-import org.apache.activemq.command.ActiveMQDestination;
-import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.activemq.command.ActiveMQTopic;
-
-/**
- * JMSInputStreamTest
- */
-@Deprecated
-public class JMSInputStreamTest extends JmsTestSupport {
-
- public Destination destination;
- protected DataOutputStream out;
- protected DataInputStream in;
- private ActiveMQConnection connection2;
-
- private ActiveMQInputStream amqIn;
- private ActiveMQOutputStream amqOut;
-
- public static Test suite() {
- return suite(JMSInputStreamTest.class);
- }
-
- public static void main(String[] args) {
- junit.textui.TestRunner.run(suite());
- }
-
- public void initCombos() {
- addCombinationValues("destination", new Object[]{new ActiveMQQueue("TEST.QUEUE"), new ActiveMQTopic("TEST.TOPIC")});
- }
-
- @Override
- protected void setUp() throws Exception {
- super.setAutoFail(true);
- super.setUp();
- }
-
- private void setUpConnection(Map<String, Object> props, long timeout) throws JMSException {
- connection2 = (ActiveMQConnection) factory.createConnection(userName, password);
- connections.add(connection2);
- if (props != null) {
- amqOut = (ActiveMQOutputStream) connection.createOutputStream(destination, props, Message.DEFAULT_DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
- }
- else {
- amqOut = (ActiveMQOutputStream) connection.createOutputStream(destination);
- }
-
- out = new DataOutputStream(amqOut);
- if (timeout == -1) {
- amqIn = (ActiveMQInputStream) connection2.createInputStream(destination);
- }
- else {
- amqIn = (ActiveMQInputStream) connection2.createInputStream(destination, null, false, timeout);
- }
- in = new DataInputStream(amqIn);
- }
-
- /*
- * @see TestCase#tearDown()
- */
- @Override
- protected void tearDown() throws Exception {
- super.tearDown();
- }
-
- /**
- * Test for AMQ-3010
- */
- public void testInputStreamTimeout() throws Exception {
- long timeout = 500;
-
- setUpConnection(null, timeout);
- try {
- in.read();
- fail();
- }
- catch (ActiveMQInputStream.ReadTimeoutException e) {
- // timeout reached, everything ok
- }
- in.close();
- }
-
- // Test for AMQ-2988
- public void testStreamsWithProperties() throws Exception {
- String name1 = "PROPERTY_1";
- String name2 = "PROPERTY_2";
- String value1 = "VALUE_1";
- String value2 = "VALUE_2";
- Map<String, Object> jmsProperties = new HashMap<>();
- jmsProperties.put(name1, value1);
- jmsProperties.put(name2, value2);
- setUpConnection(jmsProperties, -1);
-
- out.writeInt(4);
- out.flush();
- assertTrue(in.readInt() == 4);
- out.writeFloat(2.3f);
- out.flush();
- assertTrue(in.readFloat() == 2.3f);
- String str = "this is a test string";
- out.writeUTF(str);
- out.flush();
- assertTrue(in.readUTF().equals(str));
- for (int i = 0; i < 100; i++) {
- out.writeLong(i);
- }
- out.flush();
-
- // check properties before we try to read the stream
- checkProperties(jmsProperties);
-
- for (int i = 0; i < 100; i++) {
- assertTrue(in.readLong() == i);
- }
-
- // check again after read was done
- checkProperties(jmsProperties);
- }
-
- public void testStreamsWithPropertiesOnlyOnFirstMessage() throws Exception {
- String name1 = "PROPERTY_1";
- String name2 = "PROPERTY_2";
- String value1 = "VALUE_1";
- String value2 = "VALUE_2";
- Map<String, Object> jmsProperties = new HashMap<>();
- jmsProperties.put(name1, value1);
- jmsProperties.put(name2, value2);
-
- ActiveMQDestination dest = (ActiveMQDestination) destination;
-
- if (dest.isQueue()) {
- destination = new ActiveMQQueue(dest.getPhysicalName() + "?producer.addPropertiesOnFirstMsgOnly=true");
- }
- else {
- destination = new ActiveMQTopic(dest.getPhysicalName() + "?producer.addPropertiesOnFirstMsgOnly=true");
- }
-
- setUpConnection(jmsProperties, -1);
-
- assertTrue(amqOut.isAddPropertiesOnFirstMsgOnly());
-
- out.writeInt(4);
- out.flush();
- assertTrue(in.readInt() == 4);
- out.writeFloat(2.3f);
- out.flush();
- assertTrue(in.readFloat() == 2.3f);
- String str = "this is a test string";
- out.writeUTF(str);
- out.flush();
- assertTrue(in.readUTF().equals(str));
- for (int i = 0; i < 100; i++) {
- out.writeLong(i);
- }
- out.flush();
-
- // check properties before we try to read the stream
- checkProperties(jmsProperties);
-
- for (int i = 0; i < 100; i++) {
- assertTrue(in.readLong() == i);
- }
-
- // check again after read was done
- checkProperties(jmsProperties);
- }
-
- // check if the received stream has the properties set
- // Test for AMQ-2988
- private void checkProperties(Map<String, Object> jmsProperties) throws IOException {
- Map<String, Object> receivedJmsProps = amqIn.getJMSProperties();
-
- // we should at least have the same amount or more properties
- assertTrue(jmsProperties.size() <= receivedJmsProps.size());
-
- // check the properties to see if we have everything in there
- Iterator<String> propsIt = jmsProperties.keySet().iterator();
- while (propsIt.hasNext()) {
- String key = propsIt.next();
- assertTrue(receivedJmsProps.containsKey(key));
- assertEquals(jmsProperties.get(key), receivedJmsProps.get(key));
- }
- }
-
- public void testLarge() throws Exception {
- setUpConnection(null, -1);
-
- final int testData = 23;
- final int dataLength = 4096;
- final int count = 1024;
- byte[] data = new byte[dataLength];
- for (int i = 0; i < data.length; i++) {
- data[i] = testData;
- }
- final AtomicBoolean complete = new AtomicBoolean(false);
- Thread runner = new Thread(new Runnable() {
- @Override
- public void run() {
- try {
- for (int x = 0; x < count; x++) {
- byte[] b = new byte[2048];
- in.readFully(b);
- for (int i = 0; i < b.length; i++) {
- assertTrue(b[i] == testData);
- }
- }
- complete.set(true);
- synchronized (complete) {
- complete.notify();
- }
- }
- catch (Exception ex) {
- ex.printStackTrace();
- }
- }
- });
- runner.start();
- for (int i = 0; i < count; i++) {
- out.write(data);
- }
- out.flush();
- synchronized (complete) {
- while (!complete.get()) {
- complete.wait(30000);
- }
- }
- assertTrue(complete.get());
- }
-
- public void testStreams() throws Exception {
- setUpConnection(null, -1);
- out.writeInt(4);
- out.flush();
- assertTrue(in.readInt() == 4);
- out.writeFloat(2.3f);
- out.flush();
- assertTrue(in.readFloat() == 2.3f);
- String str = "this is a test string";
- out.writeUTF(str);
- out.flush();
- assertTrue(in.readUTF().equals(str));
- for (int i = 0; i < 100; i++) {
- out.writeLong(i);
- }
- out.flush();
-
- for (int i = 0; i < 100; i++) {
- assertTrue(in.readLong() == i);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/JmsResourceProvider.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/JmsResourceProvider.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/JmsResourceProvider.java
new file mode 100755
index 0000000..45be088
--- /dev/null
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/JmsResourceProvider.java
@@ -0,0 +1,258 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.test;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionConsumer;
+import javax.jms.ConnectionFactory;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.ServerSessionPool;
+import javax.jms.Session;
+import javax.jms.Topic;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+
+/**
+ *
+ */
+public class JmsResourceProvider {
+
+ private String serverUri = "vm://localhost?broker.persistent=false";
+ private boolean transacted;
+ private int ackMode = Session.AUTO_ACKNOWLEDGE;
+ private boolean isTopic;
+ private int deliveryMode = DeliveryMode.PERSISTENT;
+ private String durableName = "DummyName";
+ private String clientID = getClass().getName();
+
+ /**
+ * Creates a connection factory.
+ *
+ * @see org.apache.activemq.test.JmsResourceProvider#createConnectionFactory()
+ */
+ public ConnectionFactory createConnectionFactory() throws Exception {
+ return new ActiveMQConnectionFactory(serverUri);
+ }
+
+ /**
+ * Creates a connection.
+ *
+ * @see org.apache.activemq.test.JmsResourceProvider#createConnection(javax.jms.ConnectionFactory)
+ */
+ public Connection createConnection(ConnectionFactory cf) throws JMSException {
+ Connection connection = cf.createConnection();
+ if (getClientID() != null) {
+ connection.setClientID(getClientID());
+ }
+ return connection;
+ }
+
+ /**
+ * @see org.apache.activemq.test.JmsResourceProvider#createSession(javax.jms.Connection)
+ */
+ public Session createSession(Connection conn) throws JMSException {
+ return conn.createSession(transacted, ackMode);
+ }
+
+ /**
+ * @see org.apache.activemq.test.JmsResourceProvider#createConsumer(javax.jms.Session,
+ * javax.jms.Destination)
+ */
+ public MessageConsumer createConsumer(Session session, Destination destination) throws JMSException {
+ if (isDurableSubscriber()) {
+ return session.createDurableSubscriber((Topic)destination, durableName);
+ }
+ return session.createConsumer(destination);
+ }
+
+ /**
+ * Creates a connection for a consumer.
+ *
+ * @param ssp - ServerSessionPool
+ * @return ConnectionConsumer
+ */
+ public ConnectionConsumer createConnectionConsumer(Connection connection, Destination destination, ServerSessionPool ssp) throws JMSException {
+ return connection.createConnectionConsumer(destination, null, ssp, 1);
+ }
+
+ /**
+ * Creates a producer.
+ *
+ * @see org.apache.activemq.test.JmsResourceProvider#createProducer(javax.jms.Session,
+ * javax.jms.Destination)
+ */
+ public MessageProducer createProducer(Session session, Destination destination) throws JMSException {
+ MessageProducer producer = session.createProducer(destination);
+ producer.setDeliveryMode(deliveryMode);
+ return producer;
+ }
+
+ /**
+ * Creates a destination, which can either a topic or a queue.
+ *
+ * @see org.apache.activemq.test.JmsResourceProvider#createDestination(javax.jms.Session,
+ * String)
+ */
+ public Destination createDestination(Session session, String name) throws JMSException {
+ if (isTopic) {
+ return session.createTopic("TOPIC." + name);
+ } else {
+ return session.createQueue("QUEUE." + name);
+ }
+ }
+
+ /**
+ * Returns true if the subscriber is durable.
+ *
+ * @return isDurableSubscriber
+ */
+ public boolean isDurableSubscriber() {
+ return isTopic && durableName != null;
+ }
+
+ /**
+ * Returns the acknowledgement mode.
+ *
+ * @return Returns the ackMode.
+ */
+ public int getAckMode() {
+ return ackMode;
+ }
+
+ /**
+ * Sets the acnknowledgement mode.
+ *
+ * @param ackMode The ackMode to set.
+ */
+ public void setAckMode(int ackMode) {
+ this.ackMode = ackMode;
+ }
+
+ /**
+ * Returns true if the destination is a topic, false if the destination is a
+ * queue.
+ *
+ * @return Returns the isTopic.
+ */
+ public boolean isTopic() {
+ return isTopic;
+ }
+
+ /**
+ * @param isTopic The isTopic to set.
+ */
+ public void setTopic(boolean isTopic) {
+ this.isTopic = isTopic;
+ }
+
+ /**
+ * Returns the server URI.
+ *
+ * @return Returns the serverUri.
+ */
+ public String getServerUri() {
+ return serverUri;
+ }
+
+ /**
+ * Sets the server URI.
+ *
+ * @param serverUri - the server URI to set.
+ */
+ public void setServerUri(String serverUri) {
+ this.serverUri = serverUri;
+ }
+
+ /**
+ * Return true if the session is transacted.
+ *
+ * @return Returns the transacted.
+ */
+ public boolean isTransacted() {
+ return transacted;
+ }
+
+ /**
+ * Sets the session to be transacted.
+ *
+ * @param transacted
+ */
+ public void setTransacted(boolean transacted) {
+ this.transacted = transacted;
+ if (transacted) {
+ setAckMode(Session.SESSION_TRANSACTED);
+ }
+ }
+
+ /**
+ * Returns the delivery mode.
+ *
+ * @return deliveryMode
+ */
+ public int getDeliveryMode() {
+ return deliveryMode;
+ }
+
+ /**
+ * Sets the delivery mode.
+ *
+ * @param deliveryMode
+ */
+ public void setDeliveryMode(int deliveryMode) {
+ this.deliveryMode = deliveryMode;
+ }
+
+ /**
+ * Returns the client id.
+ *
+ * @return clientID
+ */
+ public String getClientID() {
+ return clientID;
+ }
+
+ /**
+ * Sets the client id.
+ *
+ * @param clientID
+ */
+ public void setClientID(String clientID) {
+ this.clientID = clientID;
+ }
+
+ /**
+ * Returns the durable name of the provider.
+ *
+ * @return durableName
+ */
+ public String getDurableName() {
+ return durableName;
+ }
+
+ /**
+ * Sets the durable name of the provider.
+ *
+ * @param durableName
+ */
+ public void setDurableName(String durableName) {
+ this.durableName = durableName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/TestSupport.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/TestSupport.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/TestSupport.java
new file mode 100755
index 0000000..be32877
--- /dev/null
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/test/TestSupport.java
@@ -0,0 +1,256 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.test;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.TextMessage;
+import java.io.File;
+import java.lang.reflect.Array;
+
+import junit.framework.TestCase;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Useful base class for unit test cases
+ *
+ *
+ */
+public abstract class TestSupport extends TestCase {
+ private static final Logger LOG = LoggerFactory.getLogger(TestSupport.class);
+
+ protected ActiveMQConnectionFactory connectionFactory;
+ protected boolean topic = true;
+
+ public TestSupport() {
+ super();
+ }
+
+ public TestSupport(String name) {
+ super(name);
+ }
+
+ /**
+ * Creates an ActiveMQMessage.
+ *
+ * @return ActiveMQMessage
+ */
+ protected ActiveMQMessage createMessage() {
+ return new ActiveMQMessage();
+ }
+
+ /**
+ * Creates a destination.
+ *
+ * @param subject - topic or queue name.
+ * @return Destination - either an ActiveMQTopic or ActiveMQQUeue.
+ */
+ protected Destination createDestination(String subject) {
+ if (topic) {
+ return new ActiveMQTopic(subject);
+ } else {
+ return new ActiveMQQueue(subject);
+ }
+ }
+
+ /**
+ * Tests if firstSet and secondSet are equal.
+ *
+ * @param messsage - string to be displayed when the assertion fails.
+ * @param firstSet[] - set of messages to be compared with its counterpart
+ * in the secondset.
+ * @param secondSet[] - set of messages to be compared with its counterpart
+ * in the firstset.
+ * @throws javax.jms.JMSException
+ */
+ protected void assertTextMessagesEqual(Message[] firstSet, Message[] secondSet) throws JMSException {
+ assertTextMessagesEqual("", firstSet, secondSet);
+ }
+
+ /**
+ * Tests if firstSet and secondSet are equal.
+ *
+ * @param messsage - string to be displayed when the assertion fails.
+ * @param firstSet[] - set of messages to be compared with its counterpart
+ * in the secondset.
+ * @param secondSet[] - set of messages to be compared with its counterpart
+ * in the firstset.
+ */
+ protected void assertTextMessagesEqual(String messsage, Message[] firstSet, Message[] secondSet) throws JMSException {
+ assertEquals("Message count does not match: " + messsage, firstSet.length, secondSet.length);
+
+ for (int i = 0; i < secondSet.length; i++) {
+ TextMessage m1 = (TextMessage)firstSet[i];
+ TextMessage m2 = (TextMessage)secondSet[i];
+ assertTextMessageEqual("Message " + (i + 1) + " did not match : ", m1, m2);
+ }
+ }
+
+ /**
+ * Tests if m1 and m2 are equal.
+ *
+ * @param m1 - message to be compared with m2.
+ * @param m2 - message to be compared with m1.
+ * @throws javax.jms.JMSException
+ */
+ protected void assertEquals(TextMessage m1, TextMessage m2) throws JMSException {
+ assertEquals("", m1, m2);
+ }
+
+ /**
+ * Tests if m1 and m2 are equal.
+ *
+ * @param message - string to be displayed when the assertion fails.
+ * @param m1 - message to be compared with m2.
+ * @param m2 - message to be compared with m1.
+ */
+ protected void assertTextMessageEqual(String message, TextMessage m1, TextMessage m2) throws JMSException {
+ assertFalse(message + ": expected {" + m1 + "}, but was {" + m2 + "}", m1 == null ^ m2 == null);
+
+ if (m1 == null) {
+ return;
+ }
+
+ assertEquals(message, m1.getText(), m2.getText());
+ }
+
+ /**
+ * Tests if m1 and m2 are equal.
+ *
+ * @param m1 - message to be compared with m2.
+ * @param m2 - message to be compared with m1.
+ * @throws javax.jms.JMSException
+ */
+ protected void assertEquals(Message m1, Message m2) throws JMSException {
+ assertEquals("", m1, m2);
+ }
+
+ /**
+ * Tests if m1 and m2 are equal.
+ *
+ * @param message - error message.
+ * @param m1 - message to be compared with m2.
+ * @param m2 -- message to be compared with m1.
+ */
+ protected void assertEquals(String message, Message m1, Message m2) throws JMSException {
+ assertFalse(message + ": expected {" + m1 + "}, but was {" + m2 + "}", m1 == null ^ m2 == null);
+
+ if (m1 == null) {
+ return;
+ }
+
+ assertTrue(message + ": expected {" + m1 + "}, but was {" + m2 + "}", m1.getClass() == m2.getClass());
+
+ if (m1 instanceof TextMessage) {
+ assertTextMessageEqual(message, (TextMessage)m1, (TextMessage)m2);
+ } else {
+ assertEquals(message, m1, m2);
+ }
+ }
+
+ /**
+ * Test if base directory contains spaces
+ */
+ protected void assertBaseDirectoryContainsSpaces() {
+ assertFalse("Base directory cannot contain spaces.", new File(System.getProperty("basedir", ".")).getAbsoluteFile().toString().contains(" "));
+ }
+
+ /**
+ * Creates an ActiveMQConnectionFactory.
+ *
+ * @return ActiveMQConnectionFactory
+ * @throws Exception
+ */
+ protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
+ return new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
+ }
+
+ /**
+ * Factory method to create a new connection.
+ *
+ * @return connection
+ * @throws Exception
+ */
+ protected Connection createConnection() throws Exception {
+ return getConnectionFactory().createConnection();
+ }
+
+ /**
+ * Creates an ActiveMQ connection factory.
+ *
+ * @return connectionFactory
+ * @throws Exception
+ */
+ public ActiveMQConnectionFactory getConnectionFactory() throws Exception {
+ if (connectionFactory == null) {
+ connectionFactory = createConnectionFactory();
+ assertTrue("Should have created a connection factory!", connectionFactory != null);
+ }
+
+ return connectionFactory;
+ }
+
+ /**
+ * Returns the consumer subject.
+ *
+ * @return String
+ */
+ protected String getConsumerSubject() {
+ return getSubject();
+ }
+
+ /**
+ * Returns the producer subject.
+ *
+ * @return String
+ */
+ protected String getProducerSubject() {
+ return getSubject();
+ }
+
+ /**
+ * Returns the subject.
+ *
+ * @return String
+ */
+ protected String getSubject() {
+ return getClass().getName() + "." + getName();
+ }
+
+ protected void assertArrayEqual(String message, Object[] expected, Object[] actual) {
+ assertEquals(message + ". Array length", expected.length, actual.length);
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(message + ". element: " + i, expected[i], actual[i]);
+ }
+ }
+
+ protected void assertPrimitiveArrayEqual(String message, Object expected, Object actual) {
+ int length = Array.getLength(expected);
+ assertEquals(message + ". Array length", length, Array.getLength(actual));
+ for (int i = 0; i < length; i++) {
+ assertEquals(message + ". element: " + i, Array.get(expected, i), Array.get(actual, i));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/QueueClusterTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/QueueClusterTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/QueueClusterTest.java
index 9902bd2..56d1546 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/QueueClusterTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/QueueClusterTest.java
@@ -16,13 +16,16 @@
*/
package org.apache.activemq.transport;
+import org.junit.Before;
+
/**
*
*/
public class QueueClusterTest extends TopicClusterTest {
@Override
- protected void setUp() throws Exception {
+ @Before
+ public void setUp() throws Exception {
topic = false;
super.setUp();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/SoWriteTimeoutClientTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/SoWriteTimeoutClientTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/SoWriteTimeoutClientTest.java
index 1b95006..3506ff0 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/SoWriteTimeoutClientTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/SoWriteTimeoutClientTest.java
@@ -21,42 +21,50 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
import javax.jms.Session;
-import junit.framework.Test;
-
import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.JmsTestSupport;
-import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.jms.server.config.impl.JMSConfigurationImpl;
+import org.apache.activemq.artemis.jms.server.embedded.EmbeddedJMS;
+import org.apache.activemq.broker.artemiswrapper.OpenwireArtemisBaseTest;
import org.apache.activemq.command.ActiveMQQueue;
-import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.util.SocketProxy;
-import org.apache.activemq.util.URISupport;
-import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class SoWriteTimeoutClientTest extends JmsTestSupport {
+public class SoWriteTimeoutClientTest extends OpenwireArtemisBaseTest {
private static final Logger LOG = LoggerFactory.getLogger(SoWriteTimeoutClientTest.class);
+ private String messageTextPrefix = "";
+ private EmbeddedJMS server;
+
+ @Before
+ public void setUp() throws Exception {
+ Configuration config = this.createConfig(0);
+ server = new EmbeddedJMS().setConfiguration(config).setJmsConfiguration(new JMSConfigurationImpl());
+ server.start();
+ }
- @Override
- protected BrokerService createBroker() throws Exception {
- BrokerService broker = new BrokerService();
- broker.setDeleteAllMessagesOnStartup(true);
- KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
- adapter.setConcurrentStoreAndDispatchQueues(false);
- broker.setPersistenceAdapter(adapter);
- broker.addConnector("tcp://localhost:0?wireFormat.maxInactivityDuration=0");
- return broker;
+ @After
+ public void tearDown() throws Exception {
+ server.stop();
}
+ @Test
public void testSendWithClientWriteTimeout() throws Exception {
final ActiveMQQueue dest = new ActiveMQQueue("testClientWriteTimeout");
messageTextPrefix = initMessagePrefix(80 * 1024);
- URI tcpBrokerUri = URISupport.removeQuery(broker.getTransportConnectors().get(0).getConnectUri());
+ URI tcpBrokerUri = new URI(newURI(0));
LOG.info("consuming using uri: " + tcpBrokerUri);
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(tcpBrokerUri);
@@ -92,25 +100,34 @@ public class SoWriteTimeoutClientTest extends JmsTestSupport {
TimeUnit.SECONDS.sleep(8);
proxy.goOn();
for (int i = 0; i < messageCount; i++) {
- assertNotNull("Got message " + i + " after reconnect", consumer.receive(5000));
+ Assert.assertNotNull("Got message " + i + " after reconnect", consumer.receive(10000));
}
- assertTrue("no pending messages when done", Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() throws Exception {
+ Assert.assertNull(consumer.receive(5000));
- LOG.info("current total message count: " + broker.getAdminView().getTotalMessageCount());
- return broker.getAdminView().getTotalMessageCount() == 0;
- }
- }));
+ }
+
+ protected void sendMessages(Connection connection, Destination destination, int count) throws JMSException {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ sendMessages(session, destination, count);
+ session.close();
+ }
+
+ protected void sendMessages(Session session, Destination destination, int count) throws JMSException {
+ MessageProducer producer = session.createProducer(destination);
+ sendMessages(session, producer, count);
+ producer.close();
+ }
+
+ protected void sendMessages(Session session, MessageProducer producer, int count) throws JMSException {
+ for (int i = 0; i < count; i++) {
+ producer.send(session.createTextMessage(messageTextPrefix + i));
+ }
}
private String initMessagePrefix(int i) {
byte[] content = new byte[i];
return new String(content);
}
+}
- public static Test suite() {
- return suite(SoWriteTimeoutClientTest.class);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/9307f771/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/TopicClusterTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/TopicClusterTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/TopicClusterTest.java
index 5157c33..c2a7d24 100644
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/TopicClusterTest.java
+++ b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/transport/TopicClusterTest.java
@@ -17,9 +17,6 @@
package org.apache.activemq.transport;
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
@@ -33,22 +30,23 @@ import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
-import junit.framework.TestCase;
-
import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.artemis.jms.server.embedded.EmbeddedJMS;
+import org.apache.activemq.broker.artemiswrapper.OpenwireArtemisBaseTest;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.command.ActiveMQTopic;
-import org.apache.activemq.util.ServiceStopper;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
*/
-public class TopicClusterTest extends TestCase implements MessageListener {
+public class TopicClusterTest extends OpenwireArtemisBaseTest implements MessageListener {
protected static final int MESSAGE_COUNT = 50;
protected static final int NUMBER_IN_CLUSTER = 3;
@@ -60,11 +58,11 @@ public class TopicClusterTest extends TestCase implements MessageListener {
protected int deliveryMode = DeliveryMode.NON_PERSISTENT;
protected MessageProducer[] producers;
protected Connection[] connections;
- protected List<BrokerService> services = new ArrayList<>();
+ protected EmbeddedJMS[] servers = new EmbeddedJMS[NUMBER_IN_CLUSTER];
protected String groupId;
- @Override
- protected void setUp() throws Exception {
+ @Before
+ public void setUp() throws Exception {
groupId = "topic-cluster-test-" + System.currentTimeMillis();
connections = new Connection[NUMBER_IN_CLUSTER];
producers = new MessageProducer[NUMBER_IN_CLUSTER];
@@ -73,11 +71,13 @@ public class TopicClusterTest extends TestCase implements MessageListener {
if (root == null) {
root = "target/store";
}
+
+ this.setUpClusterServers(servers);
try {
for (int i = 0; i < NUMBER_IN_CLUSTER; i++) {
System.setProperty("activemq.store.dir", root + "_broker_" + i);
- connections[i] = createConnection("broker-" + i);
+ connections[i] = createConnection(i);
connections[i].setClientID("ClusterTest" + i);
connections[i].start();
Session session = connections[i].createSession(false, Session.AUTO_ACKNOWLEDGE);
@@ -95,42 +95,35 @@ public class TopicClusterTest extends TestCase implements MessageListener {
}
}
- @Override
- protected void tearDown() throws Exception {
+ @After
+ public void tearDown() throws Exception {
if (connections != null) {
for (int i = 0; i < connections.length; i++) {
- connections[i].close();
+ try {
+ connections[i].close();
+ } catch (Exception e) {
+ //ignore.
+ }
}
}
- ServiceStopper stopper = new ServiceStopper();
- stopper.stopServices(services);
+ this.shutDownClusterServers(servers);
}
protected MessageConsumer createMessageConsumer(Session session, Destination destination) throws JMSException {
return session.createConsumer(destination);
}
- protected ActiveMQConnectionFactory createGenericClusterFactory(String brokerName) throws Exception {
- BrokerService container = new BrokerService();
- container.setBrokerName(brokerName);
-
- String url = "tcp://localhost:0";
- TransportConnector connector = container.addConnector(url);
- connector.setDiscoveryUri(new URI("multicast://default?group=" + groupId));
- container.addNetworkConnector("multicast://default?group=" + groupId);
- container.start();
-
- services.add(container);
-
- return new ActiveMQConnectionFactory("vm://" + brokerName);
+ protected ActiveMQConnectionFactory createGenericClusterFactory(int serverID) throws Exception {
+ String url = newURI(serverID);
+ return new ActiveMQConnectionFactory(url);
}
protected int expectedReceiveCount() {
return MESSAGE_COUNT * NUMBER_IN_CLUSTER * NUMBER_IN_CLUSTER;
}
- protected Connection createConnection(String name) throws Exception {
- return createGenericClusterFactory(name).createConnection();
+ protected Connection createConnection(int serverID) throws Exception {
+ return createGenericClusterFactory(serverID).createConnection();
}
protected Destination createDestination() {
@@ -146,10 +139,6 @@ public class TopicClusterTest extends TestCase implements MessageListener {
}
}
- /**
- * @param msg
- */
- @Override
public void onMessage(Message msg) {
// log.info("GOT: " + msg);
receivedMessageCount.incrementAndGet();
@@ -160,9 +149,7 @@ public class TopicClusterTest extends TestCase implements MessageListener {
}
}
- /**
- * @throws Exception
- */
+ @Test
public void testSendReceive() throws Exception {
for (int i = 0; i < MESSAGE_COUNT; i++) {
TextMessage textMessage = new ActiveMQTextMessage();
@@ -178,8 +165,8 @@ public class TopicClusterTest extends TestCase implements MessageListener {
}
// sleep a little - to check we don't get too many messages
Thread.sleep(2000);
- LOG.info("GOT: " + receivedMessageCount.get());
- assertEquals("Expected message count not correct", expectedReceiveCount(), receivedMessageCount.get());
+ LOG.info("GOT: " + receivedMessageCount.get() + " Expected: " + expectedReceiveCount());
+ Assert.assertEquals("Expected message count not correct", expectedReceiveCount(), receivedMessageCount.get());
}
}