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/21 23:55:24 UTC
[08/68] [abbrv] activemq-artemis git commit: open wire changes
equivalent to ab16f7098fb52d2b4c40627ed110e1776525f208
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCMessagePriorityTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCMessagePriorityTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCMessagePriorityTest.java
deleted file mode 100644
index ae0ac1f..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCMessagePriorityTest.java
+++ /dev/null
@@ -1,451 +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.jdbc;
-
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Vector;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import javax.jms.Connection;
-import javax.jms.DeliveryMode;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import javax.jms.TopicSubscriber;
-
-import junit.framework.Test;
-
-import org.apache.activemq.command.ActiveMQMessage;
-import org.apache.activemq.command.ActiveMQTopic;
-import org.apache.activemq.store.MessagePriorityTest;
-import org.apache.activemq.store.PersistenceAdapter;
-import org.apache.activemq.util.Wait;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class JDBCMessagePriorityTest extends MessagePriorityTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(JDBCMessagePriorityTest.class);
- EmbeddedDataSource dataSource;
- JDBCPersistenceAdapter jdbc;
-
- @Override
- protected PersistenceAdapter createPersistenceAdapter(boolean delete) throws Exception {
- jdbc = new JDBCPersistenceAdapter();
- dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- dataSource.setShutdownDatabase(null);
- jdbc.setDataSource(dataSource);
- jdbc.deleteAllMessages();
- jdbc.setCleanupPeriod(2000);
- return jdbc;
- }
-
- @Override
- protected void tearDown() throws Exception {
- super.tearDown();
- try {
- if (dataSource != null) {
- // ref http://svn.apache.org/viewvc/db/derby/code/trunk/java/testing/org/apache/derbyTesting/junit/JDBCDataSource.java?view=markup
- dataSource.setShutdownDatabase("shutdown");
- dataSource.getConnection();
- }
- }
- catch (Exception ignored) {
- }
- finally {
- dataSource.setShutdownDatabase(null);
- }
-
- }
-
- // this cannot be a general test as kahaDB just has support for 3 priority levels
- public void testDurableSubsReconnectWithFourLevels() throws Exception {
- ActiveMQTopic topic = (ActiveMQTopic) sess.createTopic("TEST");
- final String subName = "priorityDisconnect";
- TopicSubscriber sub = sess.createDurableSubscriber(topic, subName);
- sub.close();
-
- final int MED_PRI = LOW_PRI + 1;
- final int MED_HIGH_PRI = HIGH_PRI - 1;
-
- ProducerThread lowPri = new ProducerThread(topic, MSG_NUM, LOW_PRI);
- ProducerThread medPri = new ProducerThread(topic, MSG_NUM, MED_PRI);
- ProducerThread medHighPri = new ProducerThread(topic, MSG_NUM, MED_HIGH_PRI);
- ProducerThread highPri = new ProducerThread(topic, MSG_NUM, HIGH_PRI);
-
- lowPri.start();
- highPri.start();
- medPri.start();
- medHighPri.start();
-
- lowPri.join();
- highPri.join();
- medPri.join();
- medHighPri.join();
-
- final int closeFrequency = MSG_NUM;
- final int[] priorities = new int[]{HIGH_PRI, MED_HIGH_PRI, MED_PRI, LOW_PRI};
- sub = sess.createDurableSubscriber(topic, subName);
- for (int i = 0; i < MSG_NUM * 4; i++) {
- Message msg = sub.receive(10000);
- LOG.debug("received i=" + i + ", m=" + (msg != null ? msg.getJMSMessageID() + ", priority: " + msg.getJMSPriority() : null));
- assertNotNull("Message " + i + " was null", msg);
- assertEquals("Message " + i + " has wrong priority", priorities[i / MSG_NUM], msg.getJMSPriority());
- if (i > 0 && i % closeFrequency == 0) {
- LOG.info("Closing durable sub.. on: " + i);
- sub.close();
- sub = sess.createDurableSubscriber(topic, subName);
- }
- }
- LOG.info("closing on done!");
- sub.close();
- }
-
- public void initCombosForTestConcurrentDurableSubsReconnectWithXLevels() {
- addCombinationValues("prioritizeMessages", new Object[]{Boolean.TRUE, Boolean.FALSE});
- }
-
- public void testConcurrentDurableSubsReconnectWithXLevels() throws Exception {
- ActiveMQTopic topic = (ActiveMQTopic) sess.createTopic("TEST");
- final String subName = "priorityDisconnect";
- Connection consumerConn = factory.createConnection();
- consumerConn.setClientID("priorityDisconnect");
- consumerConn.start();
- Session consumerSession = consumerConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
- TopicSubscriber sub = consumerSession.createDurableSubscriber(topic, subName);
- sub.close();
-
- final int maxPriority = 5;
-
- final AtomicInteger[] messageCounts = new AtomicInteger[maxPriority];
- final long[] messageIds = new long[maxPriority];
- Vector<ProducerThread> producers = new Vector<>();
- for (int priority = 0; priority < maxPriority; priority++) {
- producers.add(new ProducerThread(topic, MSG_NUM, priority));
- messageCounts[priority] = new AtomicInteger(0);
- messageIds[priority] = 1L;
- }
-
- for (ProducerThread producer : producers) {
- producer.start();
- }
-
- final int closeFrequency = MSG_NUM / 2;
- HashMap<String, String> dups = new HashMap<>();
- sub = consumerSession.createDurableSubscriber(topic, subName);
- for (int i = 0; i < MSG_NUM * maxPriority; i++) {
- Message msg = sub.receive(10000);
- LOG.debug("received i=" + i + ", m=" + (msg != null ? msg.getJMSMessageID() + ", priority: " + msg.getJMSPriority() : null));
- assertNotNull("Message " + i + " was null, counts: " + Arrays.toString(messageCounts), msg);
- assertNull("no duplicate message failed on : " + msg.getJMSMessageID(), dups.put(msg.getJMSMessageID(), subName));
- messageCounts[msg.getJMSPriority()].incrementAndGet();
- assertEquals("message is in order : " + msg, messageIds[msg.getJMSPriority()], ((ActiveMQMessage) msg).getMessageId().getProducerSequenceId());
- messageIds[msg.getJMSPriority()]++;
- if (i > 0 && i % closeFrequency == 0) {
- LOG.info("Closing durable sub.. on: " + i + ", counts: " + Arrays.toString(messageCounts));
- sub.close();
- sub = consumerSession.createDurableSubscriber(topic, subName);
- }
- }
- LOG.info("closing on done!");
- sub.close();
- consumerSession.close();
- consumerConn.close();
-
- for (ProducerThread producer : producers) {
- producer.join();
- }
- }
-
- public void initCombosForTestConcurrentRate() {
- addCombinationValues("prefetchVal", new Object[]{new Integer(1), new Integer(500)});
- }
-
- public void testConcurrentRate() throws Exception {
- ActiveMQTopic topic = (ActiveMQTopic) sess.createTopic("TEST");
- final String subName = "priorityConcurrent";
- Connection consumerConn = factory.createConnection();
- consumerConn.setClientID("subName");
- consumerConn.start();
- Session consumerSession = consumerConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- TopicSubscriber sub = consumerSession.createDurableSubscriber(topic, subName);
- sub.close();
-
- final int TO_SEND = 2000;
- final Vector<Message> duplicates = new Vector<>();
- final int[] dups = new int[TO_SEND * 4];
- long start;
- double max = 0, sum = 0;
- MessageProducer messageProducer = sess.createProducer(topic);
- TextMessage message = sess.createTextMessage();
- for (int i = 0; i < TO_SEND; i++) {
- int priority = i % 10;
- message.setText(i + "-" + priority);
- message.setIntProperty("seq", i);
- message.setJMSPriority(priority);
- if (i > 0 && i % 1000 == 0) {
- LOG.info("Max send time: " + max + ". Sending message: " + message.getText());
- }
- start = System.currentTimeMillis();
- messageProducer.send(message, DeliveryMode.PERSISTENT, message.getJMSPriority(), 0);
- long duration = System.currentTimeMillis() - start;
- max = Math.max(max, duration);
- if (duration == max) {
- LOG.info("new max: " + max + " on i=" + i + ", " + message.getText());
- }
- sum += duration;
- }
-
- LOG.info("Sent: " + TO_SEND + ", max send time: " + max);
-
- double noConsumerAve = (sum * 100 / TO_SEND);
- sub = consumerSession.createDurableSubscriber(topic, subName);
- final AtomicInteger count = new AtomicInteger();
- sub.setMessageListener(new MessageListener() {
- @Override
- public void onMessage(Message message) {
- try {
- count.incrementAndGet();
- if (count.get() % 100 == 0) {
- LOG.info("onMessage: count: " + count.get() + ", " + ((TextMessage) message).getText() + ", seqNo " + message.getIntProperty("seq") + ", " + message.getJMSMessageID());
- }
- int seqNo = message.getIntProperty("seq");
- if (dups[seqNo] == 0) {
- dups[seqNo] = 1;
- }
- else {
- LOG.error("Duplicate: " + ((TextMessage) message).getText() + ", " + message.getJMSMessageID());
- duplicates.add(message);
- }
- }
- catch (Exception e) {
- e.printStackTrace();
- }
- }
- });
-
- LOG.info("Activated consumer");
- sum = max = 0;
- for (int i = TO_SEND; i < (TO_SEND * 2); i++) {
- int priority = i % 10;
- message.setText(i + "-" + priority);
- message.setIntProperty("seq", i);
- message.setJMSPriority(priority);
- if (i > 0 && i % 1000 == 0) {
- LOG.info("Max send time: " + max + ". Sending message: " + message.getText());
- }
- start = System.currentTimeMillis();
- messageProducer.send(message, DeliveryMode.PERSISTENT, message.getJMSPriority(), 0);
- long duration = System.currentTimeMillis() - start;
- max = Math.max(max, duration);
- if (duration == max) {
- LOG.info("new max: " + max + " on i=" + i + ", " + message.getText());
- }
- sum += duration;
- }
- LOG.info("Sent another: " + TO_SEND + ", max send time: " + max);
-
- double withConsumerAve = (sum * 100 / TO_SEND);
- final int reasonableMultiplier = 4; // not so reasonable, but on slow disks it can be
- assertTrue("max X times as slow with consumer:" + withConsumerAve + " , noConsumerMax:" + noConsumerAve, withConsumerAve < noConsumerAve * reasonableMultiplier);
- Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- LOG.info("count: " + count.get());
- return TO_SEND * 2 == count.get();
- }
- }, 60 * 1000);
-
- assertTrue("No duplicates : " + duplicates, duplicates.isEmpty());
- assertEquals("got all messages", TO_SEND * 2, count.get());
- }
-
- public void testCleanupPriorityDestination() throws Exception {
- assertEquals("no messages pending", 0, messageTableCount());
-
- ActiveMQTopic topic = (ActiveMQTopic) sess.createTopic("TEST");
- final String subName = "priorityConcurrent";
- Connection consumerConn = factory.createConnection();
- consumerConn.setClientID("subName");
- consumerConn.start();
- Session consumerSession = consumerConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- TopicSubscriber sub = consumerSession.createDurableSubscriber(topic, subName);
- sub.close();
-
- MessageProducer messageProducer = sess.createProducer(topic);
- Message message = sess.createTextMessage();
- message.setJMSPriority(2);
- messageProducer.send(message, DeliveryMode.PERSISTENT, message.getJMSPriority(), 0);
- message.setJMSPriority(5);
- messageProducer.send(message, DeliveryMode.PERSISTENT, message.getJMSPriority(), 0);
-
- assertEquals("two messages pending", 2, messageTableCount());
-
- sub = consumerSession.createDurableSubscriber(topic, subName);
-
- message = sub.receive(5000);
- assertEquals("got high priority", 5, message.getJMSPriority());
-
- waitForAck(5);
-
- for (int i = 0; i < 10; i++) {
- jdbc.cleanup();
- }
- assertEquals("one messages pending", 1, messageTableCount());
-
- message = sub.receive(5000);
- assertEquals("got high priority", 2, message.getJMSPriority());
-
- waitForAck(2);
-
- for (int i = 0; i < 10; i++) {
- jdbc.cleanup();
- }
- assertEquals("no messages pending", 0, messageTableCount());
- }
-
- public void testCleanupNonPriorityDestination() throws Exception {
- assertEquals("no messages pending", 0, messageTableCount());
-
- ActiveMQTopic topic = (ActiveMQTopic) sess.createTopic("TEST_CLEANUP_NO_PRIORITY");
- final String subName = "subName";
- Connection consumerConn = factory.createConnection();
- consumerConn.setClientID("subName");
- consumerConn.start();
- Session consumerSession = consumerConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- TopicSubscriber sub = consumerSession.createDurableSubscriber(topic, subName);
- sub.close();
-
- MessageProducer messageProducer = sess.createProducer(topic);
- Message message = sess.createTextMessage("ToExpire");
- messageProducer.send(message, DeliveryMode.PERSISTENT, Message.DEFAULT_PRIORITY, 4000);
-
- message = sess.createTextMessage("A");
- messageProducer.send(message);
- message = sess.createTextMessage("B");
- messageProducer.send(message);
- message = null;
-
- assertEquals("three messages pending", 3, messageTableCount());
-
- // let first message expire
- TimeUnit.SECONDS.sleep(5);
-
- sub = consumerSession.createDurableSubscriber(topic, subName);
- message = sub.receive(5000);
- assertNotNull("got message", message);
- LOG.info("Got: " + message);
-
- waitForAck(0, 1);
-
- for (int i = 0; i < 10; i++) {
- jdbc.cleanup();
- }
- assertEquals("one messages pending", 1, messageTableCount());
-
- message = sub.receive(5000);
- assertNotNull("got message two", message);
- LOG.info("Got: " + message);
-
- waitForAck(0, 2);
-
- for (int i = 0; i < 10; i++) {
- jdbc.cleanup();
- }
- assertEquals("no messages pending", 0, messageTableCount());
- }
-
- private int messageTableCount() throws Exception {
- int count = -1;
- java.sql.Connection c = dataSource.getConnection();
- try {
- PreparedStatement s = c.prepareStatement("SELECT COUNT(*) FROM ACTIVEMQ_MSGS");
- ResultSet rs = s.executeQuery();
- if (rs.next()) {
- count = rs.getInt(1);
- }
- }
- finally {
- if (c != null) {
- c.close();
- }
- }
- return count;
- }
-
- private void waitForAck(final int priority) throws Exception {
- waitForAck(priority, 0);
- }
-
- private void waitForAck(final int priority, final int minId) throws Exception {
- assertTrue("got ack for " + priority, Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- int id = 0;
- java.sql.Connection c = dataSource.getConnection();
- try {
- PreparedStatement s = c.prepareStatement("SELECT LAST_ACKED_ID FROM ACTIVEMQ_ACKS WHERE PRIORITY=" + priority);
- ResultSet rs = s.executeQuery();
- if (rs.next()) {
- id = rs.getInt(1);
- }
- }
- finally {
- if (c != null) {
- c.close();
- }
- }
- return id > minId;
- }
- }));
- }
-
- @SuppressWarnings("unused")
- private int messageTableDump() throws Exception {
- int count = -1;
- java.sql.Connection c = dataSource.getConnection();
- try {
- PreparedStatement s = c.prepareStatement("SELECT * FROM ACTIVEMQ_MSGS");
- ResultSet rs = s.executeQuery();
- if (rs.next()) {
- count = rs.getInt(1);
- }
- }
- finally {
- if (c != null) {
- c.close();
- }
- }
- return count;
- }
-
- public static Test suite() {
- return suite(JDBCMessagePriorityTest.class);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNegativeQueueTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNegativeQueueTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNegativeQueueTest.java
deleted file mode 100644
index e41cf13..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNegativeQueueTest.java
+++ /dev/null
@@ -1,93 +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.jdbc;
-
-import java.io.PrintStream;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.region.cursors.NegativeQueueTest;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-
-public class JDBCNegativeQueueTest extends NegativeQueueTest {
-
- EmbeddedDataSource dataSource;
-
- @Override
- protected void configureBroker(BrokerService answer) throws Exception {
- super.configureBroker(answer);
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
- answer.setPersistenceAdapter(jdbc);
- }
-
- @Override
- protected void tearDown() throws Exception {
- if (DEBUG) {
- printQuery("Select * from ACTIVEMQ_MSGS", System.out);
- }
- super.tearDown();
- }
-
- private void printQuery(String query, PrintStream out) throws SQLException {
- Connection conn = dataSource.getConnection();
- printQuery(conn.prepareStatement(query), out);
- conn.close();
- }
-
- private void printQuery(PreparedStatement s, PrintStream out) throws SQLException {
-
- ResultSet set = null;
- try {
- set = s.executeQuery();
- ResultSetMetaData metaData = set.getMetaData();
- for (int i = 1; i <= metaData.getColumnCount(); i++) {
- if (i == 1)
- out.print("||");
- out.print(metaData.getColumnName(i) + "||");
- }
- out.println();
- while (set.next()) {
- for (int i = 1; i <= metaData.getColumnCount(); i++) {
- if (i == 1)
- out.print("|");
- out.print(set.getString(i) + "|");
- }
- out.println();
- }
- }
- finally {
- try {
- set.close();
- }
- catch (Throwable ignore) {
- }
- try {
- s.close();
- }
- catch (Throwable ignore) {
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNetworkBrokerDetachTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNetworkBrokerDetachTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNetworkBrokerDetachTest.java
deleted file mode 100644
index 8e0c387..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCNetworkBrokerDetachTest.java
+++ /dev/null
@@ -1,37 +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.jdbc;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.network.NetworkBrokerDetachTest;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-
-public class JDBCNetworkBrokerDetachTest extends NetworkBrokerDetachTest {
-
- @Override
- protected void configureBroker(BrokerService broker) throws Exception {
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName(broker.getBrokerName());
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
- jdbc.deleteAllMessages();
- broker.setPersistenceAdapter(jdbc);
- broker.setUseVirtualTopics(false);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapterTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapterTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapterTest.java
deleted file mode 100644
index 59c447b..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapterTest.java
+++ /dev/null
@@ -1,67 +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.jdbc;
-
-import java.io.IOException;
-
-import junit.framework.AssertionFailedError;
-
-import org.apache.activemq.store.PersistenceAdapter;
-import org.apache.activemq.store.PersistenceAdapterTestSupport;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-
-public class JDBCPersistenceAdapterTest extends PersistenceAdapterTestSupport {
-
- @Override
- protected PersistenceAdapter createPersistenceAdapter(boolean delete) throws IOException {
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
-
- // explicitly enable audit as it is now off by default
- // due to org.apache.activemq.broker.ProducerBrokerExchange.canDispatch(Message)
- jdbc.setEnableAudit(true);
-
- brokerService.setSchedulerSupport(false);
- brokerService.setPersistenceAdapter(jdbc);
- jdbc.setBrokerService(brokerService);
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
- if (delete) {
- jdbc.deleteAllMessages();
- }
- return jdbc;
- }
-
- public void testAuditOff() throws Exception {
- pa.stop();
- pa = createPersistenceAdapter(true);
- ((JDBCPersistenceAdapter) pa).setEnableAudit(false);
- pa.start();
- boolean failed = true;
- try {
- testStoreCanHandleDupMessages();
- failed = false;
- }
- catch (AssertionFailedError e) {
- }
-
- if (!failed) {
- fail("Should have failed with audit turned off");
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreAutoCommitTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreAutoCommitTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreAutoCommitTest.java
deleted file mode 100644
index de57cbc..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreAutoCommitTest.java
+++ /dev/null
@@ -1,515 +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.jdbc;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.net.URI;
-import java.sql.Array;
-import java.sql.Blob;
-import java.sql.CallableStatement;
-import java.sql.Clob;
-import java.sql.Connection;
-import java.sql.DatabaseMetaData;
-import java.sql.NClob;
-import java.sql.PreparedStatement;
-import java.sql.SQLClientInfoException;
-import java.sql.SQLException;
-import java.sql.SQLFeatureNotSupportedException;
-import java.sql.SQLWarning;
-import java.sql.SQLXML;
-import java.sql.Savepoint;
-import java.sql.Statement;
-import java.sql.Struct;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.Executor;
-import java.util.logging.Logger;
-
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-import org.junit.Test;
-
-/**
- * to be compliant with JDBC spec; officially commit is not supposed to be
- * called on a connection that uses autocommit.The oracle v12 driver does a
- * check for autocommitSpecCompliance and it causes issues
- * <br>
- * To test; wrap the datasource used by the broker and check for autocommit
- * before delegating to real datasource. If commit is called on connection with
- * autocommit, wrapper throws a SQLException.
- */
-
-public class JDBCStoreAutoCommitTest {
-
- private static final String BROKER_NAME = "AutoCommitTest";
- private static final String TEST_DEST = "commitCheck";
- private static final String MSG_TEXT = "JDBCStoreAutoCommitTest TEST";
-
- /**
- * verify dropping and recreating tables
- *
- * @throws Exception
- */
- @Test
- public void testDeleteAllMessages() throws Exception {
- BrokerService broker = createBrokerService();
- broker.getPersistenceAdapter().deleteAllMessages();
- broker.setUseJmx(false);
- broker.start();
- broker.waitUntilStarted();
- broker.stop();
- broker.waitUntilStopped();
- }
-
- /**
- * Send message and consume message, JMS session is not transacted
- *
- * @throws Exception
- */
- @Test
- public void testSendConsume() throws Exception {
- this.doSendConsume(false);
- }
-
- /**
- * send message and consume message, JMS session is transacted
- *
- * @throws Exception
- */
- @Test
- public void testSendConsumeTransacted() throws Exception {
- this.doSendConsume(true);
- }
-
- private void doSendConsume(boolean transacted) throws Exception {
-
- BrokerService broker = createBrokerService();
- broker.setUseJmx(false);
- broker.start();
- broker.waitUntilStarted();
-
- ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(new URI("vm:" + BROKER_NAME));
- ActiveMQConnection c1 = (ActiveMQConnection) cf.createConnection();
- c1.start();
-
- try {
- // message send
- Session session1 = c1.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
- MessageProducer messageProducer = session1.createProducer(session1.createQueue(TEST_DEST));
- TextMessage textMessage = session1.createTextMessage(MSG_TEXT);
- messageProducer.send(textMessage);
-
- if (transacted) {
- session1.commit();
- }
-
- // consume
- Session session2 = c1.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
- MessageConsumer messageConsumer = session2.createConsumer(session2.createQueue(TEST_DEST));
- TextMessage messageReceived = (TextMessage) messageConsumer.receive(1000);
-
- assertEquals("check message received", MSG_TEXT, messageReceived.getText());
- }
- finally {
- c1.close();
- broker.stop();
- broker.waitUntilStopped();
- }
- }
-
- private BrokerService createBrokerService() throws IOException {
- BrokerService broker = new BrokerService();
- broker.setBrokerName(BROKER_NAME);
- broker.setUseJmx(false);
-
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource embeddedDataSource = new EmbeddedDataSource();
- embeddedDataSource.setDatabaseName("derbyDb");
- embeddedDataSource.setCreateDatabase("create");
-
- javax.sql.DataSource wrappedDataSource = new TestDataSource(embeddedDataSource);
-
- jdbc.setDataSource(wrappedDataSource);
-
- broker.setPersistenceAdapter(jdbc);
- return broker;
- }
-
- private class TestDataSource implements javax.sql.DataSource {
-
- private final javax.sql.DataSource realDataSource;
-
- public TestDataSource(javax.sql.DataSource dataSource) {
- realDataSource = dataSource;
- }
-
- @Override
- public Connection getConnection() throws SQLException {
- Connection autoCommitCheckConnection = new AutoCommitCheckConnection(realDataSource.getConnection());
- return autoCommitCheckConnection;
- }
-
- @Override
- public Connection getConnection(String username, String password) throws SQLException {
- Connection autoCommitCheckConnection = new AutoCommitCheckConnection(realDataSource.getConnection(username, password));
-
- return autoCommitCheckConnection;
- }
-
- @Override
- public PrintWriter getLogWriter() throws SQLException {
- return realDataSource.getLogWriter();
- }
-
- @Override
- public void setLogWriter(PrintWriter out) throws SQLException {
- realDataSource.setLogWriter(out);
- }
-
- @Override
- public void setLoginTimeout(int seconds) throws SQLException {
- realDataSource.setLoginTimeout(seconds);
- }
-
- @Override
- public int getLoginTimeout() throws SQLException {
- return realDataSource.getLoginTimeout();
- }
-
- @Override
- public Logger getParentLogger() throws SQLFeatureNotSupportedException {
- return realDataSource.getParentLogger();
- }
-
- @Override
- public <T> T unwrap(Class<T> iface) throws SQLException {
- return realDataSource.unwrap(iface);
- }
-
- @Override
- public boolean isWrapperFor(Class<?> iface) throws SQLException {
- return realDataSource.isWrapperFor(iface);
- }
- }
-
- private class AutoCommitCheckConnection implements Connection {
-
- private final Connection realConnection;
-
- public AutoCommitCheckConnection(Connection connection) {
- this.realConnection = connection;
- }
-
- // verify commit is not called on an auto-commit connection
- @Override
- public void commit() throws SQLException {
- if (getAutoCommit() == true) {
- throw new SQLException("AutoCommitCheckConnection: Called commit on autoCommit Connection");
- }
- realConnection.commit();
- }
-
- // Just plumbing for wrapper. Might have been better to do a Dynamic Proxy here.
-
- @Override
- public Statement createStatement() throws SQLException {
- return realConnection.createStatement();
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql) throws SQLException {
- return realConnection.prepareStatement(sql);
- }
-
- @Override
- public CallableStatement prepareCall(String sql) throws SQLException {
- return realConnection.prepareCall(sql);
- }
-
- @Override
- public String nativeSQL(String sql) throws SQLException {
- return realConnection.nativeSQL(sql);
- }
-
- @Override
- public void setAutoCommit(boolean autoCommit) throws SQLException {
- realConnection.setAutoCommit(autoCommit);
- }
-
- @Override
- public boolean getAutoCommit() throws SQLException {
- return realConnection.getAutoCommit();
- }
-
- @Override
- public void rollback() throws SQLException {
- realConnection.rollback();
- }
-
- @Override
- public void close() throws SQLException {
- realConnection.close();
- }
-
- @Override
- public boolean isClosed() throws SQLException {
- return realConnection.isClosed();
- }
-
- @Override
- public DatabaseMetaData getMetaData() throws SQLException {
- return realConnection.getMetaData();
- }
-
- @Override
- public void setReadOnly(boolean readOnly) throws SQLException {
- realConnection.setReadOnly(readOnly);
- }
-
- @Override
- public boolean isReadOnly() throws SQLException {
- return realConnection.isReadOnly();
- }
-
- @Override
- public void setCatalog(String catalog) throws SQLException {
- realConnection.setCatalog(catalog);
- }
-
- @Override
- public String getCatalog() throws SQLException {
- return realConnection.getCatalog();
- }
-
- @Override
- public void setTransactionIsolation(int level) throws SQLException {
- realConnection.setTransactionIsolation(level);
- }
-
- @Override
- public int getTransactionIsolation() throws SQLException {
- return realConnection.getTransactionIsolation();
- }
-
- @Override
- public SQLWarning getWarnings() throws SQLException {
- return realConnection.getWarnings();
- }
-
- @Override
- public void clearWarnings() throws SQLException {
- realConnection.clearWarnings();
- }
-
- @Override
- public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
- return realConnection.createStatement(resultSetType, resultSetConcurrency);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql,
- int resultSetType,
- int resultSetConcurrency) throws SQLException {
- return realConnection.prepareStatement(sql, resultSetType, resultSetConcurrency);
- }
-
- @Override
- public CallableStatement prepareCall(String sql,
- int resultSetType,
- int resultSetConcurrency) throws SQLException {
- return realConnection.prepareCall(sql, resultSetType, resultSetConcurrency);
- }
-
- @Override
- public Map<String, Class<?>> getTypeMap() throws SQLException {
- return realConnection.getTypeMap();
- }
-
- @Override
- public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
- realConnection.setTypeMap(map);
- }
-
- @Override
- public void setHoldability(int holdability) throws SQLException {
- realConnection.setHoldability(holdability);
- }
-
- @Override
- public int getHoldability() throws SQLException {
- return realConnection.getHoldability();
- }
-
- @Override
- public Savepoint setSavepoint() throws SQLException {
- return realConnection.setSavepoint();
- }
-
- @Override
- public Savepoint setSavepoint(String name) throws SQLException {
- return realConnection.setSavepoint(name);
- }
-
- @Override
- public void rollback(Savepoint savepoint) throws SQLException {
- realConnection.rollback();
- }
-
- @Override
- public void releaseSavepoint(Savepoint savepoint) throws SQLException {
- realConnection.releaseSavepoint(savepoint);
- }
-
- @Override
- public Statement createStatement(int resultSetType,
- int resultSetConcurrency,
- int resultSetHoldability) throws SQLException {
- return realConnection.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql,
- int resultSetType,
- int resultSetConcurrency,
- int resultSetHoldability) throws SQLException {
- return realConnection.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
- }
-
- @Override
- public CallableStatement prepareCall(String sql,
- int resultSetType,
- int resultSetConcurrency,
- int resultSetHoldability) throws SQLException {
- return realConnection.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
- return realConnection.prepareStatement(sql, autoGeneratedKeys);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
- return realConnection.prepareStatement(sql, columnIndexes);
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
- return realConnection.prepareStatement(sql, columnNames);
- }
-
- @Override
- public Clob createClob() throws SQLException {
- return realConnection.createClob();
- }
-
- @Override
- public Blob createBlob() throws SQLException {
- return realConnection.createBlob();
- }
-
- @Override
- public NClob createNClob() throws SQLException {
- return realConnection.createNClob();
- }
-
- @Override
- public SQLXML createSQLXML() throws SQLException {
- return realConnection.createSQLXML();
- }
-
- @Override
- public boolean isValid(int timeout) throws SQLException {
- return realConnection.isValid(timeout);
- }
-
- @Override
- public void setClientInfo(String name, String value) throws SQLClientInfoException {
- realConnection.setClientInfo(name, value);
- }
-
- @Override
- public void setClientInfo(Properties properties) throws SQLClientInfoException {
- realConnection.setClientInfo(properties);
- }
-
- @Override
- public String getClientInfo(String name) throws SQLException {
- return realConnection.getClientInfo(name);
- }
-
- @Override
- public Properties getClientInfo() throws SQLException {
- return realConnection.getClientInfo();
- }
-
- @Override
- public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
- return realConnection.createArrayOf(typeName, elements);
- }
-
- @Override
- public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
- return realConnection.createStruct(typeName, attributes);
- }
-
- @Override
- public void setSchema(String schema) throws SQLException {
- realConnection.setSchema(schema);
- }
-
- @Override
- public String getSchema() throws SQLException {
- return realConnection.getSchema();
- }
-
- @Override
- public void abort(Executor executor) throws SQLException {
- realConnection.abort(executor);
- }
-
- @Override
- public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
- realConnection.setNetworkTimeout(executor, milliseconds);
- }
-
- @Override
- public int getNetworkTimeout() throws SQLException {
- return realConnection.getNetworkTimeout();
- }
-
- @Override
- public <T> T unwrap(Class<T> iface) throws SQLException {
- return realConnection.unwrap(iface);
- }
-
- @Override
- public boolean isWrapperFor(Class<?> iface) throws SQLException {
- return realConnection.isWrapperFor(iface);
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreBrokerTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreBrokerTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreBrokerTest.java
deleted file mode 100644
index 0c86237..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreBrokerTest.java
+++ /dev/null
@@ -1,60 +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.jdbc;
-
-import junit.framework.Test;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.BrokerTest;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-
-public class JDBCStoreBrokerTest extends BrokerTest {
-
- @Override
- protected BrokerService createBroker() throws Exception {
- BrokerService broker = new BrokerService();
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
-
- jdbc.deleteAllMessages();
- broker.setPersistenceAdapter(jdbc);
- return broker;
- }
-
- protected BrokerService createRestartedBroker() throws Exception {
- BrokerService broker = new BrokerService();
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
- broker.setPersistenceAdapter(jdbc);
- return broker;
- }
-
- public static Test suite() {
- return suite(JDBCStoreBrokerTest.class);
- }
-
- public static void main(String[] args) {
- junit.textui.TestRunner.run(suite());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreOrderTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreOrderTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreOrderTest.java
deleted file mode 100644
index 17310cb..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCStoreOrderTest.java
+++ /dev/null
@@ -1,62 +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.jdbc;
-
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.command.Message;
-import org.apache.activemq.openwire.OpenWireFormat;
-import org.apache.activemq.store.StoreOrderTest;
-import org.apache.activemq.util.ByteSequence;
-import org.apache.activemq.wireformat.WireFormat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-
-// https://issues.apache.org/activemq/browse/AMQ-2594
-public class JDBCStoreOrderTest extends StoreOrderTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(JDBCStoreOrderTest.class);
-
- @Override
- protected void dumpMessages() throws Exception {
- WireFormat wireFormat = new OpenWireFormat();
- java.sql.Connection conn = ((JDBCPersistenceAdapter) broker.getPersistenceAdapter()).getDataSource().getConnection();
- PreparedStatement statement = conn.prepareStatement("SELECT ID, MSG FROM ACTIVEMQ_MSGS");
- ResultSet result = statement.executeQuery();
- while (result.next()) {
- long id = result.getLong(1);
- Message message = (Message) wireFormat.unmarshal(new ByteSequence(result.getBytes(2)));
- LOG.info("id: " + id + ", message SeqId: " + message.getMessageId().getBrokerSequenceId() + ", MSG: " + message);
- }
- statement.close();
- conn.close();
- }
-
- @Override
- protected void setPersistentAdapter(BrokerService brokerService) throws Exception {
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
- brokerService.setPersistenceAdapter(jdbc);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTablePrefixAssignedTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTablePrefixAssignedTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTablePrefixAssignedTest.java
deleted file mode 100644
index 7ac10b5..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTablePrefixAssignedTest.java
+++ /dev/null
@@ -1,133 +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.jdbc;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
-
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.util.ArrayList;
-import java.util.List;
-
-import javax.jms.Destination;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.command.Message;
-import org.apache.activemq.openwire.OpenWireFormat;
-import org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter;
-import org.apache.activemq.util.ByteSequence;
-import org.apache.activemq.wireformat.WireFormat;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class JDBCTablePrefixAssignedTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(JDBCTablePrefixAssignedTest.class);
-
- private BrokerService service;
-
- @Before
- public void setUp() throws Exception {
- service = createBroker();
- service.start();
- service.waitUntilStarted();
- }
-
- @After
- public void tearDown() throws Exception {
- service.stop();
- service.waitUntilStopped();
- }
-
- @Test
- public void testTablesHave() throws Exception {
-
- ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?create=false");
- ActiveMQConnection connection = (ActiveMQConnection) cf.createConnection();
-
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createQueue("TEST.FOO");
- MessageProducer producer = session.createProducer(destination);
-
- for (int i = 0; i < 10; ++i) {
- producer.send(session.createTextMessage("test"));
- }
- producer.close();
- connection.close();
-
- List<Message> queuedMessages = null;
- try {
- queuedMessages = dumpMessages();
- }
- catch (Exception ex) {
- LOG.info("Caught ex: ", ex);
- fail("Should not have thrown an exception");
- }
-
- assertNotNull(queuedMessages);
- assertEquals("Should have found 10 messages", 10, queuedMessages.size());
- }
-
- protected List<Message> dumpMessages() throws Exception {
- WireFormat wireFormat = new OpenWireFormat();
- java.sql.Connection conn = ((JDBCPersistenceAdapter) service.getPersistenceAdapter()).getDataSource().getConnection();
- PreparedStatement statement = conn.prepareStatement("SELECT ID, MSG FROM MYPREFIX_ACTIVEMQ_MSGS");
- ResultSet result = statement.executeQuery();
- ArrayList<Message> results = new ArrayList<>();
- while (result.next()) {
- long id = result.getLong(1);
- Message message = (Message) wireFormat.unmarshal(new ByteSequence(result.getBytes(2)));
- LOG.info("id: " + id + ", message SeqId: " + message.getMessageId().getBrokerSequenceId() + ", MSG: " + message);
- results.add(message);
- }
- statement.close();
- conn.close();
-
- return results;
- }
-
- protected BrokerService createBroker() throws Exception {
- BrokerService broker = new BrokerService();
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
-
- DefaultJDBCAdapter adapter = new DefaultJDBCAdapter();
- jdbc.setAdapter(adapter);
-
- Statements statements = new Statements();
- statements.setTablePrefix("MYPREFIX_");
- jdbc.setStatements(statements);
-
- jdbc.setUseLock(false);
- jdbc.setDataSource(dataSource);
- jdbc.deleteAllMessages();
- broker.setPersistenceAdapter(jdbc);
- return broker;
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTestMemory.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTestMemory.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTestMemory.java
deleted file mode 100644
index a8ced99..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCTestMemory.java
+++ /dev/null
@@ -1,157 +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.jdbc;
-
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-
-import junit.framework.TestCase;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-import org.junit.Ignore;
-
-public class JDBCTestMemory extends TestCase {
-
- ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
- Connection conn;
- Session sess;
- Destination dest;
-
- BrokerService broker;
-
- @Override
- protected void setUp() throws Exception {
- broker = createBroker();
- broker.start();
- broker.waitUntilStarted();
- }
-
- @Override
- protected void tearDown() throws Exception {
- broker.stop();
- }
-
- protected BrokerService createBroker() throws Exception {
- BrokerService broker = new BrokerService();
- broker.setUseJmx(true);
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
-
- jdbc.deleteAllMessages();
- broker.setPersistenceAdapter(jdbc);
- broker.addConnector("tcp://0.0.0.0:61616");
- return broker;
- }
-
- protected BrokerService createRestartedBroker() throws Exception {
- BrokerService broker = new BrokerService();
- broker.setUseJmx(true);
- JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
- EmbeddedDataSource dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc.setDataSource(dataSource);
- broker.setPersistenceAdapter(jdbc);
- broker.addConnector("tcp://0.0.0.0:61616");
- return broker;
- }
-
- public void init() throws Exception {
- conn = factory.createConnection();
- conn.start();
- sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- dest = sess.createQueue("test");
- }
-
- @Ignore("requires human input to terminate!")
- public void testRecovery() throws Exception {
- init();
- MessageProducer producer = sess.createProducer(dest);
- for (int i = 0; i < 1000; i++) {
- producer.send(sess.createTextMessage("test"));
- }
- producer.close();
- sess.close();
- conn.close();
-
- broker.stop();
- broker.waitUntilStopped();
- broker = createRestartedBroker();
- broker.start();
- broker.waitUntilStarted();
-
- init();
-
- for (int i = 0; i < 10; i++) {
- new Thread("Producer " + i) {
-
- @Override
- public void run() {
- try {
- MessageProducer producer = sess.createProducer(dest);
- for (int i = 0; i < 15000; i++) {
- producer.send(sess.createTextMessage("test"));
- if (i % 100 == 0) {
- System.out.println(getName() + " sent message " + i);
- }
- }
- producer.close();
- }
- catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- }.start();
-
- new Thread("Consumer " + i) {
-
- @Override
- public void run() {
- try {
- MessageConsumer consumer = sess.createConsumer(dest);
- for (int i = 0; i < 15000; i++) {
- consumer.receive(2000);
- if (i % 100 == 0) {
- System.out.println(getName() + " received message " + i);
- }
- }
- consumer.close();
- }
- catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- }.start();
- }
-
- // Check out JConsole
- System.in.read();
- sess.close();
- conn.close();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCXACommitExceptionTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCXACommitExceptionTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCXACommitExceptionTest.java
deleted file mode 100644
index ecc07ae..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/JDBCXACommitExceptionTest.java
+++ /dev/null
@@ -1,161 +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.jdbc;
-
-import java.io.ByteArrayOutputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-import javax.jms.Destination;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.XAConnection;
-import javax.jms.XASession;
-import javax.transaction.xa.XAException;
-import javax.transaction.xa.XAResource;
-import javax.transaction.xa.Xid;
-
-import org.apache.activemq.ActiveMQXAConnectionFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-// https://issues.apache.org/activemq/browse/AMQ-2880
-public class JDBCXACommitExceptionTest extends JDBCCommitExceptionTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(JDBCXACommitExceptionTest.class);
-
- private long txGenerator = System.currentTimeMillis();
-
- protected ActiveMQXAConnectionFactory factory;
-
- boolean onePhase = true;
-
- @Override
- public void setUp() throws Exception {
- super.setUp();
-
- factory = new ActiveMQXAConnectionFactory(connectionUri + "?jms.prefetchPolicy.all=0&jms.redeliveryPolicy.maximumRedeliveries=" + messagesExpected);
- }
-
- public void testTwoPhaseSqlException() throws Exception {
- onePhase = false;
- doTestSqlException();
- }
-
- @Override
- protected int receiveMessages(int messagesExpected) throws Exception {
- XAConnection connection = factory.createXAConnection();
- connection.start();
- XASession session = connection.createXASession();
-
- jdbc.setShouldBreak(true);
-
- // first try and receive these messages, they'll continually fail
- receiveMessages(messagesExpected, session, onePhase);
-
- jdbc.setShouldBreak(false);
-
- // now that the store is sane, try and get all the messages sent
- return receiveMessages(messagesExpected, session, onePhase);
- }
-
- protected int receiveMessages(int messagesExpected, XASession session, boolean onePhase) throws Exception {
- int messagesReceived = 0;
-
- for (int i = 0; i < messagesExpected; i++) {
- Destination destination = session.createQueue("TEST");
- MessageConsumer consumer = session.createConsumer(destination);
-
- XAResource resource = session.getXAResource();
- resource.recover(XAResource.TMSTARTRSCAN);
- resource.recover(XAResource.TMNOFLAGS);
-
- Xid tid = createXid();
-
- Message message = null;
- try {
- LOG.debug("Receiving message " + (messagesReceived + 1) + " of " + messagesExpected);
- resource.start(tid, XAResource.TMNOFLAGS);
- message = consumer.receive(2000);
- LOG.info("Received : " + message);
- resource.end(tid, XAResource.TMSUCCESS);
- if (message != null) {
- if (onePhase) {
- resource.commit(tid, true);
- }
- else {
- resource.prepare(tid);
- resource.commit(tid, false);
- }
- messagesReceived++;
- }
- }
- catch (Exception e) {
- LOG.debug("Caught exception:", e);
-
- try {
- LOG.debug("Rolling back transaction (just in case, no need to do this as it is implicit in a 1pc commit failure) " + tid);
- resource.rollback(tid);
- }
- catch (XAException ex) {
- try {
- LOG.debug("Caught exception during rollback: " + ex + " forgetting transaction " + tid);
- resource.forget(tid);
- }
- catch (XAException ex1) {
- LOG.debug("rollback/forget failed: " + ex1.errorCode);
- }
- }
- }
- finally {
- if (consumer != null) {
- consumer.close();
- }
- }
- }
- return messagesReceived;
- }
-
- public Xid createXid() throws IOException {
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream os = new DataOutputStream(baos);
- os.writeLong(++txGenerator);
- os.close();
- final byte[] bs = baos.toByteArray();
-
- return new Xid() {
- @Override
- public int getFormatId() {
- return 86;
- }
-
- @Override
- public byte[] getGlobalTransactionId() {
- return bs;
- }
-
- @Override
- public byte[] getBranchQualifier() {
- return bs;
- }
- };
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/LeaseDatabaseLockerTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/LeaseDatabaseLockerTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/LeaseDatabaseLockerTest.java
deleted file mode 100644
index 5db2c05..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/jdbc/LeaseDatabaseLockerTest.java
+++ /dev/null
@@ -1,273 +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.jdbc;
-
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.Timestamp;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.activemq.broker.AbstractLocker;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter;
-import org.apache.activemq.util.Wait;
-import org.apache.derby.jdbc.EmbeddedDataSource;
-import org.jmock.Expectations;
-import org.jmock.Mockery;
-import org.jmock.lib.legacy.ClassImposteriser;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-public class LeaseDatabaseLockerTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(LeaseDatabaseLockerTest.class);
-
- JDBCPersistenceAdapter jdbc;
- BrokerService brokerService;
- EmbeddedDataSource dataSource;
-
- @Before
- public void setUpStore() throws Exception {
- dataSource = new EmbeddedDataSource();
- dataSource.setDatabaseName("derbyDb");
- dataSource.setCreateDatabase("create");
- jdbc = new JDBCPersistenceAdapter();
- jdbc.setDataSource(dataSource);
- brokerService = new BrokerService();
- jdbc.setBrokerService(brokerService);
- jdbc.getAdapter().doCreateTables(jdbc.getTransactionContext());
- }
-
- @Test
- public void testLockInterleave() throws Exception {
-
- LeaseDatabaseLocker lockerA = new LeaseDatabaseLocker();
- lockerA.setLeaseHolderId("First");
- jdbc.setLocker(lockerA);
-
- final LeaseDatabaseLocker lockerB = new LeaseDatabaseLocker();
- lockerB.setLeaseHolderId("Second");
- jdbc.setLocker(lockerB);
- final AtomicBoolean blocked = new AtomicBoolean(true);
-
- final Connection connection = dataSource.getConnection();
- printLockTable(connection);
- lockerA.start();
- printLockTable(connection);
-
- assertTrue("First has lock", lockerA.keepAlive());
-
- final CountDownLatch lockerBStarting = new CountDownLatch(1);
- ExecutorService executor = Executors.newCachedThreadPool();
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- lockerBStarting.countDown();
- lockerB.start();
- blocked.set(false);
- printLockTable(connection);
-
- }
- catch (Exception e) {
- e.printStackTrace();
- }
- }
- });
-
- Wait.waitFor(new Wait.Condition() {
- @Override
- public boolean isSatisified() throws Exception {
- return lockerBStarting.await(1, TimeUnit.SECONDS);
- }
- });
-
- TimeUnit.MILLISECONDS.sleep(lockerB.getLockAcquireSleepInterval() / 2);
- assertTrue("B is blocked", blocked.get());
-
- assertTrue("A is good", lockerA.keepAlive());
- printLockTable(connection);
-
- lockerA.stop();
- printLockTable(connection);
-
- TimeUnit.MILLISECONDS.sleep(2 * lockerB.getLockAcquireSleepInterval());
- assertFalse("lockerB has the lock", blocked.get());
- lockerB.stop();
- printLockTable(connection);
- }
-
- @Test
- public void testLockAcquireRace() throws Exception {
-
- // build a fake lock
- final String fakeId = "Anon";
- final Connection connection = dataSource.getConnection();
- printLockTable(connection);
- PreparedStatement statement = connection.prepareStatement(jdbc.getStatements().getLeaseObtainStatement());
-
- final long now = System.currentTimeMillis();
- statement.setString(1, fakeId);
- statement.setLong(2, now + 30000);
- statement.setLong(3, now);
-
- assertEquals("we got the lease", 1, statement.executeUpdate());
- printLockTable(connection);
-
- final LeaseDatabaseLocker lockerA = new LeaseDatabaseLocker();
- lockerA.setLeaseHolderId("A");
- jdbc.setLocker(lockerA);
-
- final LeaseDatabaseLocker lockerB = new LeaseDatabaseLocker();
- lockerB.setLeaseHolderId("B");
- jdbc.setLocker(lockerB);
-
- final Set<LeaseDatabaseLocker> lockedSet = new HashSet<>();
- ExecutorService executor = Executors.newCachedThreadPool();
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- lockerA.start();
- lockedSet.add(lockerA);
- printLockTable(connection);
-
- }
- catch (Exception e) {
- e.printStackTrace();
- }
- }
- });
-
- executor.execute(new Runnable() {
- @Override
- public void run() {
- try {
- lockerB.start();
- lockedSet.add(lockerB);
- printLockTable(connection);
-
- }
- catch (Exception e) {
- e.printStackTrace();
- }
- }
- });
-
- // sleep for a bit till both are alive
- TimeUnit.SECONDS.sleep(2);
- assertTrue("no start", lockedSet.isEmpty());
- assertFalse("A is blocked", lockerA.keepAlive());
- assertFalse("B is blocked", lockerB.keepAlive());
-
- LOG.info("releasing phony lock " + fakeId);
-
- statement = connection.prepareStatement(jdbc.getStatements().getLeaseUpdateStatement());
- statement.setString(1, null);
- statement.setLong(2, 0L);
- statement.setString(3, fakeId);
- assertEquals("we released " + fakeId, 1, statement.executeUpdate());
- LOG.info("released " + fakeId);
- printLockTable(connection);
-
- TimeUnit.MILLISECONDS.sleep(AbstractLocker.DEFAULT_LOCK_ACQUIRE_SLEEP_INTERVAL);
- assertEquals("one locker started", 1, lockedSet.size());
-
- assertTrue("one isAlive", lockerA.keepAlive() || lockerB.keepAlive());
-
- LeaseDatabaseLocker winner = lockedSet.iterator().next();
- winner.stop();
- lockedSet.remove(winner);
-
- TimeUnit.MILLISECONDS.sleep(AbstractLocker.DEFAULT_LOCK_ACQUIRE_SLEEP_INTERVAL);
- assertEquals("one locker started", 1, lockedSet.size());
-
- lockedSet.iterator().next().stop();
- printLockTable(connection);
- }
-
- @Test
- public void testDiffOffsetAhead() throws Exception {
- LeaseDatabaseLocker underTest = new LeaseDatabaseLocker();
- assertTrue("when ahead of db adjustment is negative", callDiffOffset(underTest, System.currentTimeMillis() - 60000) < 0);
- }
-
- @Test
- public void testDiffOffsetBehind() throws Exception {
- LeaseDatabaseLocker underTest = new LeaseDatabaseLocker();
- assertTrue("when behind db adjustment is positive", callDiffOffset(underTest, System.currentTimeMillis() + 60000) > 0);
- }
-
- @Test
- public void testDiffIngoredIfLessthanMaxAllowableDiffFromDBTime() throws Exception {
- LeaseDatabaseLocker underTest = new LeaseDatabaseLocker();
- underTest.setMaxAllowableDiffFromDBTime(60000);
- assertEquals("no adjust when under limit", 0, callDiffOffset(underTest, System.currentTimeMillis() - 40000));
- }
-
- public long callDiffOffset(LeaseDatabaseLocker underTest, final long dbTime) throws Exception {
-
- Mockery context = new Mockery() {{
- setImposteriser(ClassImposteriser.INSTANCE);
- }};
- final Statements statements = context.mock(Statements.class);
- final JDBCPersistenceAdapter jdbcPersistenceAdapter = context.mock(JDBCPersistenceAdapter.class);
- final Connection connection = context.mock(Connection.class);
- final PreparedStatement preparedStatement = context.mock(PreparedStatement.class);
- final ResultSet resultSet = context.mock(ResultSet.class);
- final Timestamp timestamp = context.mock(Timestamp.class);
-
- context.checking(new Expectations() {{
- allowing(jdbcPersistenceAdapter).getStatements();
- will(returnValue(statements));
- allowing(jdbcPersistenceAdapter);
- allowing(statements);
- allowing(connection).prepareStatement(with(any(String.class)));
- will(returnValue(preparedStatement));
- allowing(connection);
- allowing(preparedStatement).executeQuery();
- will(returnValue(resultSet));
- allowing(resultSet).next();
- will(returnValue(true));
- allowing(resultSet).getTimestamp(1);
- will(returnValue(timestamp));
- allowing(timestamp).getTime();
- will(returnValue(dbTime));
- }});
-
- underTest.configure(jdbcPersistenceAdapter);
- underTest.setLockable(jdbcPersistenceAdapter);
- return underTest.determineTimeDifference(connection);
- }
-
- private void printLockTable(Connection connection) throws Exception {
- DefaultJDBCAdapter.printQuery(connection, "SELECT * from ACTIVEMQ_LOCK", System.err);
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/4c717ca5/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/CustomLockerTest.java
----------------------------------------------------------------------
diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/CustomLockerTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/CustomLockerTest.java
deleted file mode 100644
index 8136eb6..0000000
--- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/store/kahadb/CustomLockerTest.java
+++ /dev/null
@@ -1,32 +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;
-
-import junit.framework.TestCase;
-
-import org.apache.activemq.broker.BrokerFactory;
-import org.apache.activemq.broker.BrokerService;
-
-public class CustomLockerTest extends TestCase {
-
- public void testCustomLocker() throws Exception {
- BrokerService broker = BrokerFactory.createBroker("xbean:org/apache/activemq/store/kahadb/shared.xml");
- broker.waitUntilStarted();
- broker.stop();
- broker.waitUntilStopped();
- }
-}