qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kw...@apache.org
Subject [2/2] qpid-broker-j git commit: QPID-8070:[Broker-J][JDBC Store] Instantiate asynchronous commits executor on open of JDBC message store
Date Tue, 16 Jan 2018 15:46:32 GMT
QPID-8070:[Broker-J][JDBC Store] Instantiate asynchronous commits executor on open of JDBC
message store

Cherry picked from 14a53f619628570426466655ef2aa1edac9085a3


Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/31d81ffe
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/31d81ffe
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/31d81ffe

Branch: refs/heads/6.1.x
Commit: 31d81ffe24b7c928bc855186258f74a8cf890a9a
Parents: 852aa0b
Author: Alex Rudyy <orudyy@apache.org>
Authored: Mon Jan 15 17:50:08 2018 +0000
Committer: Keith Wall <kwall@apache.org>
Committed: Tue Jan 16 15:44:28 2018 +0000

----------------------------------------------------------------------
 .../jdbc/GenericAbstractJDBCMessageStore.java   |  1 +
 .../server/store/jdbc/JDBCMessageStoreTest.java | 67 ++++++++++++++++++++
 2 files changed, 68 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/31d81ffe/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericAbstractJDBCMessageStore.java
----------------------------------------------------------------------
diff --git a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericAbstractJDBCMessageStore.java
b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericAbstractJDBCMessageStore.java
index d46ec4d..bbfe88a 100644
--- a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericAbstractJDBCMessageStore.java
+++ b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericAbstractJDBCMessageStore.java
@@ -43,6 +43,7 @@ public abstract class GenericAbstractJDBCMessageStore extends org.apache.qpid.se
         {
             _parent = parent;
             super.setTablePrefix(getTablePrefix(parent));
+            initMessageStore(parent);
             doOpen(parent);
 
             createOrOpenMessageStoreDatabase();

http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/31d81ffe/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
----------------------------------------------------------------------
diff --git a/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
b/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
index a3f9271..2bd51c6 100644
--- a/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
+++ b/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
@@ -31,10 +31,20 @@ import java.sql.SQLException;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
+import com.google.common.util.concurrent.ListenableFuture;
+
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.internal.InternalMessage;
 import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.store.MessageDurability;
+import org.apache.qpid.server.store.MessageEnqueueRecord;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.MessageStoreTestCase;
+import org.apache.qpid.server.store.Transaction;
+import org.apache.qpid.server.store.TransactionLogResource;
 import org.apache.qpid.server.virtualhost.jdbc.JDBCVirtualHost;
 
 public class JDBCMessageStoreTest extends MessageStoreTestCase
@@ -78,6 +88,63 @@ public class JDBCMessageStoreTest extends MessageStoreTestCase
         assertTablesExist(expectedTables, false);
     }
 
+    public void testEnqueueTransactionCommitAsync() throws Exception
+    {
+        final String queueName = getTestName();
+        final UUID transactionalLogId = UUID.randomUUID();
+
+        final MessageStore store = getStore();
+        final TransactionLogResource transactionalLog = mockTransactionLogResource(transactionalLogId,
queueName);
+        final InternalMessage message = addTestMessage(store, "test");
+
+        final Transaction transaction = store.newTransaction();
+        final MessageEnqueueRecord record = transaction.enqueueMessage(transactionalLog,
message);
+
+        assertNotNull("Message enqueue record should not be null", record);
+        assertEquals("Unexpected queue id", transactionalLogId, record.getQueueId());
+        assertEquals("Unexpected message number", message.getMessageNumber(), record.getMessageNumber());
+
+        final ListenableFuture<Void> future = transaction.commitTranAsync(null);
+        future.get(1000, TimeUnit.MILLISECONDS);
+    }
+
+    public void testDequeueTransactionCommitAsync() throws Exception
+    {
+        final String queueName = getTestName();
+        final UUID transactionalLogId = UUID.randomUUID();
+
+        final MessageStore store = getStore();
+        final TransactionLogResource transactionalLog = mockTransactionLogResource(transactionalLogId,
queueName);
+        final InternalMessage message = addTestMessage(store, "test2");
+
+        final Transaction enqueueTransaction = store.newTransaction();
+        MessageEnqueueRecord record = enqueueTransaction.enqueueMessage(transactionalLog,
message);
+        enqueueTransaction.commitTran();
+
+        final Transaction dequeueTransaction = store.newTransaction();
+        dequeueTransaction.dequeueMessage(record);
+
+        final ListenableFuture<Void> future = dequeueTransaction.commitTranAsync(null);
+        future.get(1000, TimeUnit.MILLISECONDS);
+    }
+
+    private InternalMessage addTestMessage(final MessageStore store,
+                                           final String messageContent)
+    {
+        final AMQMessageHeader amqpHeader = mock(AMQMessageHeader.class);
+        return InternalMessage.createMessage(store, amqpHeader, messageContent, true);
+    }
+
+    private TransactionLogResource mockTransactionLogResource(final UUID transactionalLogId,
+                                                              final String transactionalLogName)
+    {
+        final TransactionLogResource transactionalLog = mock(TransactionLogResource.class);
+        when(transactionalLog.getId()).thenReturn(transactionalLogId);
+        when(transactionalLog.getName()).thenReturn(transactionalLogName);
+        when(transactionalLog.getMessageDurability()).thenReturn(MessageDurability.ALWAYS);
+        return transactionalLog;
+    }
+
     @Override
     protected VirtualHost createVirtualHost()
     {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message