You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2017/06/19 15:00:27 UTC
[1/2] qpid-broker-j git commit: QPID-7827: Delegate the creation of
UUIDs to factory to avoid proliferation of UUID instances representing the
same underlying UUID.
Repository: qpid-broker-j
Updated Branches:
refs/heads/master 2f918544a -> 973c1f8db
QPID-7827: Delegate the creation of UUIDs to factory to avoid proliferation of UUID instances representing the same underlying UUID.
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/2ddbca0b
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/2ddbca0b
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/2ddbca0b
Branch: refs/heads/master
Commit: 2ddbca0b16e47b18f24beaa1f4fa88d594a25618
Parents: 2f91854
Author: Keith Wall <kw...@apache.org>
Authored: Mon Jun 19 14:30:22 2017 +0100
Committer: Keith Wall <kw...@apache.org>
Committed: Mon Jun 19 14:32:39 2017 +0100
----------------------------------------------------------------------
.../berkeleydb/AbstractBDBMessageStore.java | 28 ++++++------
.../tuple/PreparedTransactionBinding.java | 46 +++++++++++--------
.../berkeleydb/tuple/QueueEntryBinding.java | 47 ++++++++-----------
.../qpid/server/util/CachingUUIDFactory.java | 48 ++++++++++++++++++++
.../server/util/CachingUUIDFactoryTest.java | 47 +++++++++++++++++++
.../store/jdbc/AbstractJDBCMessageStore.java | 12 +++--
6 files changed, 161 insertions(+), 67 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2ddbca0b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
----------------------------------------------------------------------
diff --git a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
index b6bd5c9..7260557 100644
--- a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
+++ b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
@@ -71,6 +71,7 @@ import org.apache.qpid.server.store.berkeleydb.tuple.XidBinding;
import org.apache.qpid.server.store.handler.DistributedTransactionHandler;
import org.apache.qpid.server.store.handler.MessageHandler;
import org.apache.qpid.server.store.handler.MessageInstanceHandler;
+import org.apache.qpid.server.util.CachingUUIDFactory;
public abstract class AbstractBDBMessageStore implements MessageStore
@@ -624,9 +625,8 @@ public abstract class AbstractBDBMessageStore implements MessageStore
{
DatabaseEntry key = new DatabaseEntry();
- QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
- QueueEntryKey dd = new QueueEntryKey(queue.getId(), messageId);
- keyBinding.objectToEntry(dd, key);
+ QueueEntryKey queueEntryKey = new QueueEntryKey(queue.getId(), messageId);
+ QueueEntryBinding.objectToEntry(queueEntryKey, key);
DatabaseEntry value = new DatabaseEntry();
value.setData(ENQUEUE_RECORD_VALUE, 0, ENQUEUE_RECORD_VALUE.length);
@@ -666,10 +666,9 @@ public abstract class AbstractBDBMessageStore implements MessageStore
{
DatabaseEntry key = new DatabaseEntry();
- QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
QueueEntryKey queueEntryKey = new QueueEntryKey(queueId, messageId);
UUID id = queueId;
- keyBinding.objectToEntry(queueEntryKey, key);
+ QueueEntryBinding.objectToEntry(queueEntryKey, key);
getLogger().debug("Dequeue message id {} from queue with id {}", messageId, id);
@@ -713,8 +712,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore
DatabaseEntry value = new DatabaseEntry();
PreparedTransaction preparedTransaction = new PreparedTransaction(enqueues, dequeues);
- PreparedTransactionBinding valueBinding = new PreparedTransactionBinding();
- valueBinding.objectToEntry(preparedTransaction, value);
+ PreparedTransactionBinding.objectToEntry(preparedTransaction, value);
for(org.apache.qpid.server.store.Transaction.EnqueueRecord enqueue : enqueues)
{
StoredMessage storedMessage = enqueue.getMessage().getStoredMessage();
@@ -1528,13 +1526,13 @@ public abstract class AbstractBDBMessageStore implements MessageStore
DatabaseEntry value = new DatabaseEntry();
value.setPartial(0, 0, true);
- QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
- keyBinding.objectToEntry(new QueueEntryKey(queue.getId(), 0l), key);
+ CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
+ QueueEntryBinding.objectToEntry(new QueueEntryKey(queue.getId(), 0L), key);
if (!searchCompletedSuccessfully && (searchCompletedSuccessfully =
cursor.getSearchKeyRange(key, value, LockMode.READ_UNCOMMITTED) == OperationStatus.SUCCESS))
{
- QueueEntryKey entry = keyBinding.entryToObject(key);
+ QueueEntryKey entry = QueueEntryBinding.entryToObject(uuidFactory, key);
if (entry.getQueueId().equals(queue.getId()))
{
entries.add(entry);
@@ -1545,7 +1543,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore
{
while (cursor.getNext(key, value, LockMode.READ_UNCOMMITTED) == OperationStatus.SUCCESS)
{
- QueueEntryKey entry = keyBinding.entryToObject(key);
+ QueueEntryKey entry = QueueEntryBinding.entryToObject(uuidFactory, key);
if (entry.getQueueId().equals(queue.getId()))
{
entries.add(entry);
@@ -1585,13 +1583,13 @@ public abstract class AbstractBDBMessageStore implements MessageStore
try(Cursor cursor = getDeliveryDb().openCursor(null, null))
{
DatabaseEntry key = new DatabaseEntry();
- QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
+ CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
DatabaseEntry value = new DatabaseEntry();
value.setPartial(0, 0, true);
while (cursor.getNext(key, value, LockMode.READ_UNCOMMITTED) == OperationStatus.SUCCESS)
{
- QueueEntryKey entry = keyBinding.entryToObject(key);
+ QueueEntryKey entry = QueueEntryBinding.entryToObject(uuidFactory, key);
entries.add(entry);
}
}
@@ -1619,15 +1617,15 @@ public abstract class AbstractBDBMessageStore implements MessageStore
try(Cursor cursor = getXidDb().openCursor(null, null))
{
+ CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
DatabaseEntry key = new DatabaseEntry();
XidBinding keyBinding = XidBinding.getInstance();
- PreparedTransactionBinding valueBinding = new PreparedTransactionBinding();
DatabaseEntry value = new DatabaseEntry();
while (cursor.getNext(key, value, LockMode.READ_UNCOMMITTED) == OperationStatus.SUCCESS)
{
Xid xid = keyBinding.entryToObject(key);
- PreparedTransaction preparedTransaction = valueBinding.entryToObject(value);
+ PreparedTransaction preparedTransaction = PreparedTransactionBinding.entryToObject(uuidFactory, value);
if (!handler.handle(new BDBStoredXidRecord(xid.getFormat(), xid.getGlobalId(), xid.getBranchId()),
preparedTransaction.getEnqueues(), preparedTransaction.getDequeues()))
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2ddbca0b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/PreparedTransactionBinding.java
----------------------------------------------------------------------
diff --git a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/PreparedTransactionBinding.java b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/PreparedTransactionBinding.java
index 4d111e5..23e302a 100644
--- a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/PreparedTransactionBinding.java
+++ b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/PreparedTransactionBinding.java
@@ -26,6 +26,8 @@ import java.util.UUID;
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
+import com.sleepycat.je.DatabaseEntry;
+
import org.apache.qpid.server.message.EnqueueableMessage;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
@@ -34,49 +36,53 @@ import org.apache.qpid.server.store.Transaction;
import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.server.store.berkeleydb.AbstractBDBMessageStore;
import org.apache.qpid.server.store.berkeleydb.entry.PreparedTransaction;
+import org.apache.qpid.server.util.CachingUUIDFactory;
-public class PreparedTransactionBinding extends TupleBinding<PreparedTransaction>
+public class PreparedTransactionBinding
{
- @Override
- public PreparedTransaction entryToObject(TupleInput input)
+ private PreparedTransactionBinding()
{
- Transaction.EnqueueRecord[] enqueues = readEnqueueRecords(input);
-
- Transaction.DequeueRecord[] dequeues = readDequeueRecords(input);
+ }
+ public static PreparedTransaction entryToObject(final CachingUUIDFactory uuidFactory, final DatabaseEntry value)
+ {
+ TupleInput input = TupleBinding.entryToInput(value);
+ Transaction.EnqueueRecord[] enqueues = readEnqueueRecords(uuidFactory, input);
+ Transaction.DequeueRecord[] dequeues = readDequeueRecords(uuidFactory, input);
return new PreparedTransaction(enqueues, dequeues);
}
- private Transaction.EnqueueRecord[] readEnqueueRecords(TupleInput input)
+ private static Transaction.EnqueueRecord[] readEnqueueRecords(final CachingUUIDFactory uuidFactory, TupleInput input)
{
Transaction.EnqueueRecord[] records = new Transaction.EnqueueRecord[input.readInt()];
for(int i = 0; i < records.length; i++)
{
- records[i] = new EnqueueRecordImpl(new UUID(input.readLong(), input.readLong()), input.readLong());
+ UUID queueId = uuidFactory.createUuidFromBits(input.readLong(), input.readLong());
+ records[i] = new EnqueueRecordImpl(queueId, input.readLong());
}
return records;
}
- private Transaction.DequeueRecord[] readDequeueRecords(TupleInput input)
+ private static Transaction.DequeueRecord[] readDequeueRecords(final CachingUUIDFactory uuidFactory, TupleInput input)
{
Transaction.DequeueRecord[] records = new Transaction.DequeueRecord[input.readInt()];
for(int i = 0; i < records.length; i++)
{
- records[i] = new DequeueRecordImpl(new UUID(input.readLong(), input.readLong()), input.readLong());
+ UUID queueId = uuidFactory.createUuidFromBits(input.readLong(), input.readLong());
+ records[i] = new DequeueRecordImpl(queueId, input.readLong());
}
return records;
}
-
- @Override
- public void objectToEntry(PreparedTransaction preparedTransaction, TupleOutput output)
+ public static void objectToEntry(final PreparedTransaction preparedTransaction, final DatabaseEntry value)
{
- writeRecords(preparedTransaction.getEnqueues(), output);
- writeRecords(preparedTransaction.getDequeues(), output);
-
+ TupleOutput tupleOutput = new TupleOutput();
+ writeRecords(preparedTransaction.getEnqueues(), tupleOutput);
+ writeRecords(preparedTransaction.getDequeues(), tupleOutput);
+ TupleBinding.outputToEntry(tupleOutput, value);
}
- private void writeRecords(Transaction.EnqueueRecord[] records, TupleOutput output)
+ private static void writeRecords(Transaction.EnqueueRecord[] records, TupleOutput output)
{
if(records == null)
{
@@ -95,7 +101,7 @@ public class PreparedTransactionBinding extends TupleBinding<PreparedTransaction
}
}
- private void writeRecords(Transaction.DequeueRecord[] records, TupleOutput output)
+ private static void writeRecords(Transaction.DequeueRecord[] records, TupleOutput output)
{
if(records == null)
{
@@ -120,7 +126,7 @@ public class PreparedTransactionBinding extends TupleBinding<PreparedTransaction
private long _messageNumber;
private UUID _queueId;
- public EnqueueRecordImpl(UUID queueId, long messageNumber)
+ EnqueueRecordImpl(UUID queueId, long messageNumber)
{
_messageNumber = messageNumber;
_queueId = queueId;
@@ -175,7 +181,7 @@ public class PreparedTransactionBinding extends TupleBinding<PreparedTransaction
private final AbstractBDBMessageStore.BDBEnqueueRecord _record;
- public DequeueRecordImpl(final UUID queueId, final long messageNumber)
+ DequeueRecordImpl(final UUID queueId, final long messageNumber)
{
_record = new AbstractBDBMessageStore.BDBEnqueueRecord(queueId, messageNumber);
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2ddbca0b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/QueueEntryBinding.java
----------------------------------------------------------------------
diff --git a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/QueueEntryBinding.java b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/QueueEntryBinding.java
index ee00e5d..32f1bb9 100644
--- a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/QueueEntryBinding.java
+++ b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/QueueEntryBinding.java
@@ -22,46 +22,39 @@ package org.apache.qpid.server.store.berkeleydb.tuple;
import java.util.UUID;
-import com.sleepycat.bind.EntryBinding;
import com.sleepycat.je.DatabaseEntry;
import org.apache.qpid.server.store.berkeleydb.entry.QueueEntryKey;
+import org.apache.qpid.server.util.CachingUUIDFactory;
-public class QueueEntryBinding implements EntryBinding<QueueEntryKey>
+public class QueueEntryBinding
{
-
- private static final QueueEntryBinding INSTANCE = new QueueEntryBinding();
-
- public static QueueEntryBinding getInstance()
+ private QueueEntryBinding()
{
- return INSTANCE;
}
- /** private constructor forces getInstance instead */
- private QueueEntryBinding() { }
-
- public QueueEntryKey entryToObject(DatabaseEntry entry)
+ public static QueueEntryKey entryToObject(final CachingUUIDFactory uuidFactory, DatabaseEntry entry)
{
byte[] data = entry.getData();
int offset = entry.getOffset();
- UUID queueId = new UUID(readUnsignedLong(data,offset)^ 0x8000000000000000L, readUnsignedLong(data,offset+8)^ 0x8000000000000000L);
+ UUID queueId = uuidFactory.createUuidFromBits(readUnsignedLong(data, offset) ^ 0x8000000000000000L, readUnsignedLong(data, offset + 8) ^ 0x8000000000000000L);
long messageId = readUnsignedLong(data,offset+16)^ 0x8000000000000000L;
return new QueueEntryKey(queueId, messageId);
}
- public void objectToEntry(QueueEntryKey mk, DatabaseEntry entry)
+ public static void objectToEntry(QueueEntryKey entryKey, DatabaseEntry entry)
{
byte[] output = new byte[24];
- UUID uuid = mk.getQueueId();
+ UUID uuid = entryKey.getQueueId();
writeUnsignedLong(uuid.getMostSignificantBits() ^ 0x8000000000000000L, output, 0);
writeUnsignedLong(uuid.getLeastSignificantBits() ^ 0x8000000000000000L, output, 8);
- writeUnsignedLong(mk.getMessageId() ^ 0x8000000000000000L, output, 16);
+ writeUnsignedLong(entryKey.getMessageId() ^ 0x8000000000000000L, output, 16);
entry.setData(output);
}
- private void writeUnsignedLong(long val, byte[] data, int offset)
+ private static void writeUnsignedLong(long val, byte[] data, int offset)
{
data[offset++] = (byte) (val >>> 56);
data[offset++] = (byte) (val >>> 48);
@@ -73,19 +66,15 @@ public class QueueEntryBinding implements EntryBinding<QueueEntryKey>
data[offset] = (byte) val;
}
- private long readUnsignedLong(final byte[] data, int offset)
+ private static long readUnsignedLong(final byte[] data, int offset)
{
- return (((long)data[offset++] & 0xffl) << 56)
- | (((long)data[offset++] & 0xffl) << 48)
- | (((long)data[offset++] & 0xffl) << 40)
- | (((long)data[offset++] & 0xffl) << 32)
- | (((long)data[offset++] & 0xffl) << 24)
- | (((long)data[offset++] & 0xffl) << 16)
- | (((long)data[offset++] & 0xffl) << 8)
- | ((long)data[offset] & 0xffl) ;
+ return (((long)data[offset++] & 0xffL) << 56)
+ | (((long)data[offset++] & 0xffL) << 48)
+ | (((long)data[offset++] & 0xffL) << 40)
+ | (((long)data[offset++] & 0xffL) << 32)
+ | (((long)data[offset++] & 0xffL) << 24)
+ | (((long)data[offset++] & 0xffL) << 16)
+ | (((long)data[offset++] & 0xffL) << 8)
+ | ((long)data[offset] & 0xffL) ;
}
-
-
-
-
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2ddbca0b/broker-core/src/main/java/org/apache/qpid/server/util/CachingUUIDFactory.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/util/CachingUUIDFactory.java b/broker-core/src/main/java/org/apache/qpid/server/util/CachingUUIDFactory.java
new file mode 100644
index 0000000..9dd01c6
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/util/CachingUUIDFactory.java
@@ -0,0 +1,48 @@
+/*
+ * 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.qpid.server.util;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+public class CachingUUIDFactory
+{
+ private final Map<UUID, UUID> _uuids = new HashMap<>();
+
+ public UUID createUuidFromString(final String name)
+ {
+ UUID candidate = UUID.fromString(name);
+ return cacheIfNecessary(candidate);
+ }
+
+ public UUID createUuidFromBits(final long mostSigBits, final long leastSigBits)
+ {
+ UUID candidate = new UUID(mostSigBits, leastSigBits);
+ return cacheIfNecessary(candidate);
+ }
+
+ private UUID cacheIfNecessary(final UUID candidate)
+ {
+ UUID existing = _uuids.putIfAbsent(candidate, candidate);
+ return existing == null ? candidate : existing;
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2ddbca0b/broker-core/src/test/java/org/apache/qpid/server/util/CachingUUIDFactoryTest.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/util/CachingUUIDFactoryTest.java b/broker-core/src/test/java/org/apache/qpid/server/util/CachingUUIDFactoryTest.java
new file mode 100644
index 0000000..f27df9a
--- /dev/null
+++ b/broker-core/src/test/java/org/apache/qpid/server/util/CachingUUIDFactoryTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.qpid.server.util;
+
+import java.util.UUID;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class CachingUUIDFactoryTest extends QpidTestCase
+{
+ private final CachingUUIDFactory _factory = new CachingUUIDFactory();
+
+ public void testUuidFromBits()
+ {
+ UUID first = _factory.createUuidFromBits(0L,0L);
+ UUID second = _factory.createUuidFromBits(0L,0L);
+ assertSame("UUIDFactory should return the same object", first, second);
+ }
+
+ public void testUuidFromString()
+ {
+ String uuidStr = UUID.randomUUID().toString();
+ UUID first = _factory.createUuidFromString(new String(uuidStr));
+ UUID second = _factory.createUuidFromString(new String(uuidStr));
+ UUID third = _factory.createUuidFromBits(second.getMostSignificantBits(), second.getLeastSignificantBits());
+ assertSame("UUIDFactory should return the same object", first, second);
+ assertSame("UUIDFactory should return the same object", first, third);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2ddbca0b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
----------------------------------------------------------------------
diff --git a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
index 8b0a7ad..ddd5763 100644
--- a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
+++ b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
@@ -68,6 +68,7 @@ import org.apache.qpid.server.store.handler.DistributedTransactionHandler;
import org.apache.qpid.server.store.handler.MessageHandler;
import org.apache.qpid.server.store.handler.MessageInstanceHandler;
import org.apache.qpid.server.txn.Xid;
+import org.apache.qpid.server.util.CachingUUIDFactory;
public abstract class AbstractJDBCMessageStore implements MessageStore
{
@@ -1845,6 +1846,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
Connection conn = null;
try
{
+ CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
conn = newAutoCommitConnection();
PreparedStatement stmt = conn.prepareStatement("SELECT queue_id, message_id FROM " + getQueueEntryTableName()
+ " WHERE queue_id = ? ORDER BY queue_id, message_id");
@@ -1858,7 +1860,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
String id = rs.getString(1);
long messageId = rs.getLong(2);
- if (!handler.handle(new JDBCEnqueueRecord(UUID.fromString(id), messageId)))
+ UUID uuid = uuidFactory.createUuidFromString(id);
+ if (!handler.handle(new JDBCEnqueueRecord(uuid, messageId)))
{
break;
}
@@ -1893,6 +1896,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
Connection conn = null;
try
{
+ CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
conn = newAutoCommitConnection();
Statement stmt = conn.createStatement();
try
@@ -1905,7 +1909,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
String id = rs.getString(1);
long messageId = rs.getLong(2);
- if (!handler.handle(new JDBCEnqueueRecord(UUID.fromString(id), messageId)))
+ UUID queueId = uuidFactory.createUuidFromString(id);
+ if (!handler.handle(new JDBCEnqueueRecord(queueId, messageId)))
{
break;
}
@@ -1970,6 +1975,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
for (Xid xid : xids)
{
+ CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
List<RecordImpl> enqueues = new ArrayList<>();
List<RecordImpl> dequeues = new ArrayList<>();
@@ -1990,7 +1996,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
String actionType = rs.getString(1);
- UUID queueId = UUID.fromString(rs.getString(2));
+ UUID queueId = uuidFactory.createUuidFromString(rs.getString(2));
long messageId = rs.getLong(3);
RecordImpl record = new RecordImpl(queueId, messageId);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/2] qpid-broker-j git commit: QPID-7831: Refactor
AbstractJDBCMessageStore to use try-with-resources universally
Posted by kw...@apache.org.
QPID-7831: Refactor AbstractJDBCMessageStore to use try-with-resources universally
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/973c1f8d
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/973c1f8d
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/973c1f8d
Branch: refs/heads/master
Commit: 973c1f8dba1374f9f882e6ee9029891e36e8c4d2
Parents: 2ddbca0
Author: Keith Wall <kw...@apache.org>
Authored: Mon Jun 19 15:12:25 2017 +0100
Committer: Keith Wall <kw...@apache.org>
Committed: Mon Jun 19 16:00:03 2017 +0100
----------------------------------------------------------------------
.../store/jdbc/AbstractJDBCMessageStore.java | 323 +++++--------------
1 file changed, 73 insertions(+), 250 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/973c1f8d/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
----------------------------------------------------------------------
diff --git a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
index ddd5763..f831399 100644
--- a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
+++ b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
@@ -127,40 +127,26 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
private void setMaxMessageId(final Connection conn, final String query, int col) throws SQLException
{
- PreparedStatement statement =
- conn.prepareStatement(query);
- try
+ try (PreparedStatement statement = conn.prepareStatement(query))
{
- ResultSet rs = statement.executeQuery();
- try
+ try (ResultSet rs = statement.executeQuery())
{
- while(rs.next())
+ while (rs.next())
{
long maxMessageId = rs.getLong(col);
- if(_messageId.get() < maxMessageId)
+ if (_messageId.get() < maxMessageId)
{
_messageId.set(maxMessageId);
}
}
-
- }
- finally
- {
- rs.close();
}
}
- finally
- {
- statement.close();
- }
}
protected void upgrade(ConfiguredObject<?> parent) throws StoreException
{
- Connection conn = null;
- try
+ try(Connection conn = newAutoCommitConnection())
{
- conn = newAutoCommitConnection();
if (tableExists(getDbVersionTableName(), conn))
{
upgradeIfNecessary(parent);
@@ -170,10 +156,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
throw new StoreException("Failed to upgrade database", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
private void upgradeIfNecessary(ConfiguredObject<?> parent) throws SQLException
@@ -275,11 +257,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
protected void createOrOpenMessageStoreDatabase() throws StoreException
{
- Connection conn = null;
- try
+ try(Connection conn = newAutoCommitConnection())
{
- conn = newAutoCommitConnection();
-
createVersionTable(conn);
createQueueEntryTable(conn);
createMetaDataTable(conn);
@@ -291,10 +270,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
throw new StoreException("Failed to create message store tables", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
private void createVersionTable(final Connection conn) throws SQLException
@@ -444,34 +419,34 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
private void removeMessage(long messageId)
{
- try
+ try(Connection conn = newConnection())
{
- Connection conn = newConnection();
try
{
- PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getMetaDataTableName()
- + " WHERE message_id = ?");
- try
+ try(PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getMetaDataTableName()
+ + " WHERE message_id = ?"))
{
- stmt.setLong(1,messageId);
+ stmt.setLong(1, messageId);
int results = stmt.executeUpdate();
stmt.close();
if (results == 0)
{
- getLogger().debug("Message id {} not found (attempt to remove failed - probably application initiated rollback)", messageId);
+ getLogger().debug(
+ "Message id {} not found (attempt to remove failed - probably application initiated rollback)",
+
+ messageId);
}
getLogger().debug("Deleted metadata for message {}", messageId);
-
- stmt = conn.prepareStatement("DELETE FROM " + getMessageContentTableName()
- + " WHERE message_id = ?");
- stmt.setLong(1, messageId);
- results = stmt.executeUpdate();
}
- finally
+
+ try(PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getMessageContentTableName()
+ + " WHERE message_id = ?"))
{
- stmt.close();
+
+ stmt.setLong(1, messageId);
+ int results = stmt.executeUpdate();
}
conn.commit();
}
@@ -487,11 +462,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
throw e;
-
- }
- finally
- {
- conn.close();
}
}
catch (SQLException e)
@@ -636,12 +606,10 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
Connection conn = connWrapper.getConnection();
-
try
{
- PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getXidTableName()
- + " WHERE format = ? and global_id = ? and branch_id = ?");
- try
+ try(PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getXidTableName()
+ + " WHERE format = ? and global_id = ? and branch_id = ?"))
{
stmt.setLong(1,format);
stmt.setBytes(2,globalId);
@@ -655,14 +623,9 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
throw new StoreException("Unable to find message with xid");
}
}
- finally
- {
- stmt.close();
- }
- stmt = conn.prepareStatement("DELETE FROM " + getXidActionsTableName()
- + " WHERE format = ? and global_id = ? and branch_id = ?");
- try
+ try(PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getXidActionsTableName()
+ + " WHERE format = ? and global_id = ? and branch_id = ?"))
{
stmt.setLong(1,format);
stmt.setBytes(2,globalId);
@@ -670,11 +633,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
int results = stmt.executeUpdate();
}
- finally
- {
- stmt.close();
- }
-
}
catch (SQLException e)
{
@@ -693,19 +651,14 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
try
{
- PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getXidTableName()
- + " ( format, global_id, branch_id ) values (?, ?, ?)");
- try
+ try(PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getXidTableName()
+ + " ( format, global_id, branch_id ) values (?, ?, ?)"))
{
stmt.setLong(1,format);
stmt.setBytes(2, globalId);
stmt.setBytes(3, branchId);
stmt.executeUpdate();
}
- finally
- {
- stmt.close();
- }
for(Transaction.EnqueueRecord enqueue : enqueues)
{
@@ -716,12 +669,9 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
}
-
- stmt = conn.prepareStatement("INSERT INTO " + getXidActionsTableName()
- + " ( format, global_id, branch_id, action_type, " +
- "queue_id, message_id ) values (?,?,?,?,?,?) ");
-
- try
+ try(PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getXidActionsTableName()
+ + " ( format, global_id, branch_id, action_type, " +
+ "queue_id, message_id ) values (?,?,?,?,?,?) "))
{
stmt.setLong(1,format);
stmt.setBytes(2, globalId);
@@ -750,10 +700,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
}
- finally
- {
- stmt.close();
- }
return Collections.emptyList();
}
catch (SQLException e)
@@ -885,9 +831,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
getLogger().debug("Adding metadata for message {}", messageId);
- PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getMetaDataTableName()
- + "( message_id , meta_data ) values (?, ?)");
- try
+ try(PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getMetaDataTableName()
+ + "( message_id , meta_data ) values (?, ?)"))
{
stmt.setLong(1, messageId);
@@ -899,8 +844,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
buf = buf.slice();
metaData.writeToBuffer(buf);
- ByteArrayInputStream bis = new ByteArrayInputStream(underlying);
- try
+ try(ByteArrayInputStream bis = new ByteArrayInputStream(underlying))
{
stmt.setBinaryStream(2, bis, underlying.length);
int result = stmt.executeUpdate();
@@ -910,23 +854,10 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
throw new StoreException("Unable to add meta data for message " + messageId);
}
}
- finally
+ catch (IOException e)
{
- try
- {
- bis.close();
- }
- catch (IOException e)
- {
-
- throw new SQLException(e);
- }
+ throw new SQLException("Failed to close ByteArrayInputStream", e);
}
-
- }
- finally
- {
- stmt.close();
}
}
@@ -1004,25 +935,23 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
private StorableMessageMetaData getMetaData(long messageId) throws SQLException
{
- Connection conn = newAutoCommitConnection();
- try
+ try (Connection conn = newAutoCommitConnection())
{
- PreparedStatement stmt = conn.prepareStatement("SELECT meta_data FROM " + getMetaDataTableName()
- + " WHERE message_id = ?");
- try
+ try (PreparedStatement stmt = conn.prepareStatement("SELECT meta_data FROM " + getMetaDataTableName()
+ + " WHERE message_id = ?"))
{
- stmt.setLong(1,messageId);
- ResultSet rs = stmt.executeQuery();
- try
+ stmt.setLong(1, messageId);
+ try (ResultSet rs = stmt.executeQuery())
{
- if(rs.next())
+ if (rs.next())
{
byte[] dataAsBytes = getBlobAsBytes(rs, 1);
QpidByteBuffer buf = QpidByteBuffer.wrap(dataAsBytes);
buf.position(1);
buf = buf.slice();
- int typeOrdinal = dataAsBytes[0] & 0xff;;
+ int typeOrdinal = dataAsBytes[0] & 0xff;
+
MessageMetaDataType type = MessageMetaDataTypeRegistry.fromOrdinal(typeOrdinal);
StorableMessageMetaData metaData = type.createMetaData(buf);
buf.dispose();
@@ -1033,20 +962,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
throw new StoreException("Meta data not found for message with id " + messageId);
}
}
- finally
- {
- rs.close();
- }
- }
- finally
- {
- stmt.close();
}
}
- finally
- {
- conn.close();
- }
}
protected abstract byte[] getBlobAsBytes(ResultSet rs, int col) throws SQLException;
@@ -1056,8 +973,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
getLogger().debug("Adding content for message {}", messageId);
- PreparedStatement stmt = null;
-
int size = 0;
for(QpidByteBuffer buf : contentBody)
@@ -1071,11 +986,9 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
buf.copyTo(dst);
}
- try
+ try(PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getMessageContentTableName()
+ + "( message_id, content ) values (?, ?)"))
{
-
- stmt = conn.prepareStatement("INSERT INTO " + getMessageContentTableName()
- + "( message_id, content ) values (?, ?)");
stmt.setLong(1, messageId);
stmt.setBinaryStream(2, new ByteArrayInputStream(data), data.length);
stmt.executeUpdate();
@@ -1085,25 +998,16 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
JdbcUtils.closeConnection(conn, getLogger());
throw new StoreException("Error adding content for message " + messageId + ": " + e.getMessage(), e);
}
- finally
- {
- JdbcUtils.closePreparedStatement(stmt, getLogger());
- }
}
Collection<QpidByteBuffer> getAllContent(long messageId) throws StoreException
{
- Connection conn = null;
- PreparedStatement stmt = null;
-
getLogger().debug("Message Id: {} Getting content body", messageId);
- try
+ try(Connection conn = newAutoCommitConnection();
+ PreparedStatement stmt = conn.prepareStatement("SELECT content FROM " + getMessageContentTableName()
+ + " WHERE message_id = ?"))
{
- conn = newAutoCommitConnection();
-
- stmt = conn.prepareStatement("SELECT content FROM " + getMessageContentTableName()
- + " WHERE message_id = ?");
stmt.setLong(1,messageId);
ResultSet rs = stmt.executeQuery();
@@ -1132,11 +1036,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
throw new StoreException("Error retrieving content for message " + messageId + ": " + e.getMessage(), e);
}
- finally
- {
- JdbcUtils.closePreparedStatement(stmt, getLogger());
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
@Override
@@ -1737,11 +1636,9 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
checkMessageStoreOpen();
- Connection conn = null;
StoredJDBCMessage message;
- try
+ try(Connection conn = newAutoCommitConnection())
{
- conn = newAutoCommitConnection();
try (PreparedStatement stmt = conn.prepareStatement("SELECT message_id, meta_data FROM " + getMetaDataTableName()
+ " WHERE message_id = ?"))
{
@@ -1772,10 +1669,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
throw new StoreException("Error encountered when visiting messages", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
@Override
@@ -1790,15 +1683,12 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
checkMessageStoreOpen();
- Connection conn = null;
- try
+ try(Connection conn = newAutoCommitConnection())
{
- conn = newAutoCommitConnection();
- Statement stmt = conn.createStatement();
- try
+ try (Statement stmt = conn.createStatement())
{
- ResultSet rs = stmt.executeQuery("SELECT message_id, meta_data FROM " + getMetaDataTableName());
- try
+ try (ResultSet rs = stmt.executeQuery("SELECT message_id, meta_data FROM "
+ + getMetaDataTableName()))
{
while (rs.next())
{
@@ -1807,7 +1697,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
QpidByteBuffer buf = QpidByteBuffer.wrap(dataAsBytes);
buf.position(1);
buf = buf.slice();
- MessageMetaDataType<?> type = MessageMetaDataTypeRegistry.fromOrdinal(((int)dataAsBytes[0]) &0xff);
+ MessageMetaDataType<?> type =
+ MessageMetaDataTypeRegistry.fromOrdinal(((int) dataAsBytes[0]) & 0xff);
StorableMessageMetaData metaData = type.createMetaData(buf);
buf.dispose();
StoredJDBCMessage message = createStoredJDBCMessage(messageId, metaData, true);
@@ -1817,24 +1708,12 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
}
}
- finally
- {
- rs.close();
- }
- }
- finally
- {
- stmt.close();
}
}
catch (SQLException e)
{
throw new StoreException("Error encountered when visiting messages", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
@Override
@@ -1843,18 +1722,16 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
checkMessageStoreOpen();
- Connection conn = null;
- try
+ try(Connection conn = newAutoCommitConnection())
{
CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
- conn = newAutoCommitConnection();
- PreparedStatement stmt = conn.prepareStatement("SELECT queue_id, message_id FROM " + getQueueEntryTableName()
- + " WHERE queue_id = ? ORDER BY queue_id, message_id");
- try
+ try (PreparedStatement stmt = conn.prepareStatement("SELECT queue_id, message_id FROM "
+ + getQueueEntryTableName()
+ + " WHERE queue_id = ? ORDER BY queue_id, "
+ + "message_id"))
{
stmt.setString(1, queue.getId().toString());
- ResultSet rs = stmt.executeQuery();
- try
+ try (ResultSet rs = stmt.executeQuery())
{
while (rs.next())
{
@@ -1867,25 +1744,12 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
}
}
- finally
- {
- rs.close();
- }
- }
- finally
- {
- stmt.close();
}
}
catch (SQLException e)
{
throw new StoreException("Error encountered when visiting message instances", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
-
}
@Override
@@ -1893,17 +1757,13 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
checkMessageStoreOpen();
- Connection conn = null;
- try
+ try(Connection conn = newAutoCommitConnection())
{
CachingUUIDFactory uuidFactory = new CachingUUIDFactory();
- conn = newAutoCommitConnection();
- Statement stmt = conn.createStatement();
- try
+ try (Statement stmt = conn.createStatement())
{
- ResultSet rs = stmt.executeQuery("SELECT queue_id, message_id FROM " + getQueueEntryTableName()
- + " ORDER BY queue_id, message_id");
- try
+ try (ResultSet rs = stmt.executeQuery("SELECT queue_id, message_id FROM " + getQueueEntryTableName()
+ + " ORDER BY queue_id, message_id"))
{
while (rs.next())
{
@@ -1916,24 +1776,12 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
}
}
- finally
- {
- rs.close();
- }
- }
- finally
- {
- stmt.close();
}
}
catch (SQLException e)
{
throw new StoreException("Error encountered when visiting message instances", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
@Override
@@ -1941,17 +1789,14 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
checkMessageStoreOpen();
- Connection conn = null;
- try
+ try(Connection conn = newAutoCommitConnection())
{
- conn = newAutoCommitConnection();
List<Xid> xids = new ArrayList<Xid>();
- Statement stmt = conn.createStatement();
- try
+ try (Statement stmt = conn.createStatement())
{
- ResultSet rs = stmt.executeQuery("SELECT format, global_id, branch_id FROM " + getXidTableName());
- try
+ try (ResultSet rs = stmt.executeQuery("SELECT format, global_id, branch_id FROM "
+ + getXidTableName()))
{
while (rs.next())
{
@@ -1962,14 +1807,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
xids.add(new Xid(format, globalId, branchId));
}
}
- finally
- {
- rs.close();
- }
- }
- finally
- {
- stmt.close();
}
@@ -1979,18 +1816,16 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
List<RecordImpl> enqueues = new ArrayList<>();
List<RecordImpl> dequeues = new ArrayList<>();
- PreparedStatement pstmt = conn.prepareStatement("SELECT action_type, queue_id, message_id FROM " + getXidActionsTableName()
- +
- " WHERE format = ? and global_id = ? and branch_id = ?");
-
- try
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "SELECT action_type, queue_id, message_id FROM " + getXidActionsTableName()
+ +
+ " WHERE format = ? and global_id = ? and branch_id = ?"))
{
pstmt.setLong(1, xid.getFormat());
pstmt.setBytes(2, xid.getGlobalId());
pstmt.setBytes(3, xid.getBranchId());
- ResultSet rs = pstmt.executeQuery();
- try
+ try (ResultSet rs = pstmt.executeQuery())
{
while (rs.next())
{
@@ -2004,14 +1839,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
records.add(record);
}
}
- finally
- {
- rs.close();
- }
- }
- finally
- {
- pstmt.close();
}
if (!handler.handle(new JDBCStoredXidRecord(xid.getFormat(), xid.getGlobalId(), xid.getBranchId()),
@@ -2028,10 +1855,6 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
throw new StoreException("Error encountered when visiting distributed transactions", e);
}
- finally
- {
- JdbcUtils.closeConnection(conn, getLogger());
- }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org