| /* |
| * |
| * 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.store; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.sql.Connection; |
| import java.sql.PreparedStatement; |
| import java.sql.ResultSet; |
| import java.sql.SQLException; |
| import java.sql.Statement; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledThreadPoolExecutor; |
| import java.util.concurrent.ThreadFactory; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| |
| import com.google.common.util.concurrent.Futures; |
| import com.google.common.util.concurrent.ListenableFuture; |
| import com.google.common.util.concurrent.SettableFuture; |
| import org.slf4j.Logger; |
| |
| import org.apache.qpid.bytebuffer.QpidByteBuffer; |
| import org.apache.qpid.server.message.EnqueueableMessage; |
| import org.apache.qpid.server.model.ConfiguredObject; |
| import org.apache.qpid.server.plugin.MessageMetaDataType; |
| 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 AbstractJDBCMessageStore implements MessageStore |
| { |
| private static final String DB_VERSION_TABLE_NAME_SUFFIX = "QPID_DB_VERSION"; |
| private static final String QUEUE_ENTRY_TABLE_NAME_SUFFIX = "QPID_QUEUE_ENTRIES"; |
| private static final String META_DATA_TABLE_NAME_SUFFIX = "QPID_MESSAGE_METADATA"; |
| private static final String MESSAGE_CONTENT_TABLE_NAME_SUFFIX = "QPID_MESSAGE_CONTENT"; |
| private static final String XID_TABLE_NAME_SUFFIX = "QPID_XIDS"; |
| private static final String XID_ACTIONS_TABLE_NAME_SUFFIX = "QPID_XID_ACTIONS"; |
| |
| private static final int DB_VERSION = 8; |
| |
| private final AtomicLong _messageId = new AtomicLong(0); |
| |
| |
| protected final EventManager _eventManager = new EventManager(); |
| private ConfiguredObject<?> _parent; |
| private String _tablePrefix = ""; |
| private final AtomicLong _bytesEvacuatedFromMemory = new AtomicLong(); |
| private final Set<StoredJDBCMessage<?>> _messages = Collections.newSetFromMap(new ConcurrentHashMap<StoredJDBCMessage<?>, Boolean>()); |
| |
| protected abstract boolean isMessageStoreOpen(); |
| |
| protected abstract void checkMessageStoreOpen(); |
| private ScheduledThreadPoolExecutor _executor; |
| |
| public AbstractJDBCMessageStore() |
| { |
| } |
| |
| protected void setMaximumMessageId() |
| { |
| |
| try |
| { |
| Connection conn = newAutoCommitConnection(); |
| try |
| { |
| setMaxMessageId(conn, "SELECT max(message_id) FROM " + getMessageContentTableName(), 1); |
| setMaxMessageId(conn, "SELECT max(message_id) FROM " + getMetaDataTableName(), 1); |
| setMaxMessageId(conn, "SELECT queue_id, max(message_id) FROM " + getQueueEntryTableName() |
| + " GROUP BY queue_id " , 2); |
| } |
| finally |
| { |
| conn.close(); |
| } |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Failed to determine maximum ids", e); |
| } |
| |
| } |
| |
| private void setMaxMessageId(final Connection conn, final String query, int col) throws SQLException |
| { |
| PreparedStatement statement = |
| conn.prepareStatement(query); |
| try |
| { |
| ResultSet rs = statement.executeQuery(); |
| try |
| { |
| while(rs.next()) |
| { |
| long maxMessageId = rs.getLong(col); |
| if(_messageId.get() < maxMessageId) |
| { |
| _messageId.set(maxMessageId); |
| } |
| } |
| |
| } |
| finally |
| { |
| rs.close(); |
| } |
| } |
| finally |
| { |
| statement.close(); |
| } |
| } |
| |
| protected void upgrade(ConfiguredObject<?> parent) throws StoreException |
| { |
| Connection conn = null; |
| try |
| { |
| conn = newAutoCommitConnection(); |
| if (tableExists(getDbVersionTableName(), conn)) |
| { |
| upgradeIfNecessary(parent); |
| } |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Failed to upgrade database", e); |
| } |
| finally |
| { |
| JdbcUtils.closeConnection(conn, getLogger()); |
| } |
| } |
| |
| private void upgradeIfNecessary(ConfiguredObject<?> parent) throws SQLException |
| { |
| try (Connection conn = newAutoCommitConnection()) |
| { |
| |
| try (PreparedStatement statement = conn.prepareStatement("SELECT version FROM " + getDbVersionTableName())) |
| { |
| try (ResultSet rs = statement.executeQuery()) |
| { |
| if (!rs.next()) |
| { |
| throw new StoreException(getDbVersionTableName() + " does not contain the database version"); |
| } |
| int version = rs.getInt(1); |
| switch (version) |
| { |
| case 6: |
| upgradeFromV6(); |
| case 7: |
| upgradeFromV7(); |
| case DB_VERSION: |
| return; |
| default: |
| throw new StoreException("Unknown database version: " + version); |
| } |
| } |
| } |
| } |
| |
| } |
| |
| private void upgradeFromV7() throws SQLException |
| { |
| updateDbVersion(8); |
| } |
| |
| private void upgradeFromV6() throws SQLException |
| { |
| updateDbVersion(7); |
| } |
| |
| private void updateDbVersion(int newVersion) throws SQLException |
| { |
| try (Connection conn = newAutoCommitConnection()) |
| { |
| try (PreparedStatement statement = conn.prepareStatement("UPDATE " + getDbVersionTableName() |
| + " SET version = ?")) |
| { |
| statement.setInt(1, newVersion); |
| statement.execute(); |
| } |
| } |
| } |
| |
| protected void initMessageStore(final ConfiguredObject<?> parent) |
| { |
| _parent = parent; |
| _executor = new ScheduledThreadPoolExecutor(4, new ThreadFactory() |
| { |
| private final AtomicInteger _count = new AtomicInteger(); |
| @Override |
| public Thread newThread(final Runnable r) |
| { |
| final Thread thread = Executors.defaultThreadFactory().newThread(r); |
| thread.setName(parent.getName() + "-store-"+_count.incrementAndGet()); |
| return thread; |
| } |
| }); |
| _executor.prestartAllCoreThreads(); |
| |
| } |
| |
| @Override |
| public void closeMessageStore() |
| { |
| for (StoredJDBCMessage<?> message : _messages) |
| { |
| message.clear(); |
| } |
| _messages.clear(); |
| _bytesEvacuatedFromMemory.set(0); |
| if(_executor != null) |
| { |
| _executor.shutdown(); |
| } |
| |
| } |
| |
| protected abstract Logger getLogger(); |
| |
| protected abstract String getSqlBlobType(); |
| |
| protected abstract String getSqlVarBinaryType(int size); |
| |
| protected abstract String getSqlBigIntType(); |
| |
| protected void createOrOpenMessageStoreDatabase() throws StoreException |
| { |
| Connection conn = null; |
| try |
| { |
| conn = newAutoCommitConnection(); |
| |
| createVersionTable(conn); |
| createQueueEntryTable(conn); |
| createMetaDataTable(conn); |
| createMessageContentTable(conn); |
| createXidTable(conn); |
| createXidActionTable(conn); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Failed to create message store tables", e); |
| } |
| finally |
| { |
| JdbcUtils.closeConnection(conn, getLogger()); |
| } |
| } |
| |
| private void createVersionTable(final Connection conn) throws SQLException |
| { |
| if(!tableExists(getDbVersionTableName(), conn)) |
| { |
| try (Statement stmt = conn.createStatement()) |
| { |
| stmt.execute("CREATE TABLE " + getDbVersionTableName() + " ( version int not null )"); |
| } |
| |
| try (PreparedStatement pstmt = conn.prepareStatement("INSERT INTO " + getDbVersionTableName() |
| + " ( version ) VALUES ( ? )")) |
| { |
| pstmt.setInt(1, DB_VERSION); |
| pstmt.execute(); |
| } |
| } |
| } |
| |
| private void createQueueEntryTable(final Connection conn) throws SQLException |
| { |
| if(!tableExists(getQueueEntryTableName(), conn)) |
| { |
| try (Statement stmt = conn.createStatement()) |
| { |
| stmt.execute("CREATE TABLE " + getQueueEntryTableName() |
| + " ( queue_id varchar(36) not null, message_id " |
| + getSqlBigIntType() + " not null, PRIMARY KEY (queue_id, message_id) )"); |
| } |
| } |
| |
| } |
| |
| private void createMetaDataTable(final Connection conn) throws SQLException |
| { |
| if(!tableExists(getMetaDataTableName(), conn)) |
| { |
| try (Statement stmt = conn.createStatement()) |
| { |
| stmt.execute("CREATE TABLE " |
| + getMetaDataTableName() |
| + " ( message_id " |
| + getSqlBigIntType() |
| + " not null, meta_data " |
| + getSqlBlobType() |
| + ", PRIMARY KEY ( message_id ) )"); |
| } |
| } |
| |
| } |
| |
| private void createMessageContentTable(final Connection conn) throws SQLException |
| { |
| if(!tableExists(getMessageContentTableName(), conn)) |
| { |
| try (Statement stmt = conn.createStatement()) |
| { |
| stmt.execute("CREATE TABLE " |
| + getMessageContentTableName() |
| + " ( message_id " |
| + getSqlBigIntType() |
| + " not null, content " |
| + getSqlBlobType() |
| + ", PRIMARY KEY (message_id) )"); |
| } |
| } |
| |
| } |
| |
| private void createXidTable(final Connection conn) throws SQLException |
| { |
| if(!tableExists(getXidTableName(), conn)) |
| { |
| try (Statement stmt = conn.createStatement()) |
| { |
| stmt.execute("CREATE TABLE " |
| + getXidTableName() |
| + " ( format " + getSqlBigIntType() + " not null," |
| + " global_id " |
| + getSqlVarBinaryType(64) |
| + ", branch_id " |
| + getSqlVarBinaryType(64) |
| + " , PRIMARY KEY ( format, " |
| + |
| "global_id, branch_id ))"); |
| } |
| } |
| } |
| |
| private void createXidActionTable(final Connection conn) throws SQLException |
| { |
| if(!tableExists(getXidActionsTableName(), conn)) |
| { |
| try (Statement stmt = conn.createStatement()) |
| { |
| stmt.execute("CREATE TABLE " |
| + getXidActionsTableName() |
| + " ( format " |
| + getSqlBigIntType() |
| + " not null," |
| + " global_id " |
| + getSqlVarBinaryType(64) |
| + " not null, branch_id " |
| + getSqlVarBinaryType( |
| 64) |
| + " not null, " |
| + |
| "action_type char not null, queue_id varchar(36) not null, message_id " |
| + getSqlBigIntType() |
| + " not null" |
| + |
| ", PRIMARY KEY ( " |
| + |
| "format, global_id, branch_id, action_type, queue_id, message_id))"); |
| } |
| } |
| } |
| |
| protected boolean tableExists(final String tableName, final Connection conn) throws SQLException |
| { |
| return JdbcUtils.tableExists(tableName, conn); |
| } |
| |
| @Override |
| public <T extends StorableMessageMetaData> MessageHandle<T> addMessage(T metaData) |
| { |
| checkMessageStoreOpen(); |
| |
| return createStoredJDBCMessage(getNextMessageId(), metaData, false); |
| } |
| |
| public <T extends StorableMessageMetaData> StoredJDBCMessage<T> createStoredJDBCMessage(final long newMessageId, |
| final T metaData, |
| final boolean recovered) |
| { |
| final StoredJDBCMessage<T> message = new StoredJDBCMessage<>(newMessageId, metaData, recovered); |
| _messages.add(message); |
| return message; |
| } |
| |
| @Override |
| public long getNextMessageId() |
| { |
| return _messageId.incrementAndGet(); |
| } |
| |
| private void removeMessage(long messageId) |
| { |
| try |
| { |
| Connection conn = newConnection(); |
| try |
| { |
| PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getMetaDataTableName() |
| + " WHERE message_id = ?"); |
| try |
| { |
| 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("Deleted metadata for message {}", messageId); |
| |
| stmt = conn.prepareStatement("DELETE FROM " + getMessageContentTableName() |
| + " WHERE message_id = ?"); |
| stmt.setLong(1, messageId); |
| results = stmt.executeUpdate(); |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| conn.commit(); |
| } |
| catch(SQLException e) |
| { |
| try |
| { |
| conn.rollback(); |
| } |
| catch(SQLException t) |
| { |
| // ignore - we are re-throwing underlying exception |
| } |
| |
| throw e; |
| |
| } |
| finally |
| { |
| conn.close(); |
| } |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error removing message with id " + messageId + " from database: " + e.getMessage(), e); |
| } |
| |
| } |
| |
| /** |
| * Convenience method to create a new Connection configured for TRANSACTION_READ_COMMITED |
| * isolation and with auto-commit transactions enabled. |
| */ |
| protected Connection newAutoCommitConnection() throws SQLException |
| { |
| final Connection connection = newConnection(); |
| try |
| { |
| connection.setAutoCommit(true); |
| } |
| catch (SQLException sqlEx) |
| { |
| |
| try |
| { |
| connection.close(); |
| } |
| finally |
| { |
| throw sqlEx; |
| } |
| } |
| |
| return connection; |
| } |
| |
| /** |
| * Convenience method to create a new Connection configured for TRANSACTION_READ_COMMITED |
| * isolation and with auto-commit transactions disabled. |
| */ |
| protected Connection newConnection() throws SQLException |
| { |
| final Connection connection = getConnection(); |
| try |
| { |
| connection.setAutoCommit(false); |
| connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED); |
| } |
| catch (SQLException sqlEx) |
| { |
| try |
| { |
| connection.close(); |
| } |
| finally |
| { |
| throw sqlEx; |
| } |
| } |
| return connection; |
| } |
| |
| protected abstract Connection getConnection() throws SQLException; |
| |
| @Override |
| public Transaction newTransaction() |
| { |
| checkMessageStoreOpen(); |
| |
| return new JDBCTransaction(); |
| } |
| |
| private void enqueueMessage(ConnectionWrapper connWrapper, final TransactionLogResource queue, Long messageId) throws StoreException |
| { |
| Connection conn = connWrapper.getConnection(); |
| |
| |
| try |
| { |
| if (getLogger().isDebugEnabled()) |
| { |
| getLogger().debug("Enqueuing message {} on queue {} with id {} [Connection {}]", |
| messageId, queue.getName(), queue.getId(), conn); |
| } |
| |
| try (PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getQueueEntryTableName() |
| + " (queue_id, message_id) values (?,?)")) |
| { |
| stmt.setString(1, queue.getId().toString()); |
| stmt.setLong(2, messageId); |
| stmt.executeUpdate(); |
| } |
| |
| } |
| catch (SQLException e) |
| { |
| getLogger().error("Failed to enqueue message {}", messageId, e); |
| throw new StoreException("Error writing enqueued message with id " + messageId + " for queue " + queue.getName() + " with id " + queue.getId() |
| + " to database", e); |
| } |
| |
| } |
| |
| private void dequeueMessage(ConnectionWrapper connWrapper, final UUID queueId, |
| Long messageId) throws StoreException |
| { |
| |
| Connection conn = connWrapper.getConnection(); |
| |
| try |
| { |
| try (PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getQueueEntryTableName() |
| + " WHERE queue_id = ? AND message_id =?")) |
| { |
| stmt.setString(1, queueId.toString()); |
| stmt.setLong(2, messageId); |
| int results = stmt.executeUpdate(); |
| |
| |
| if (results != 1) |
| { |
| throw new StoreException("Unable to find message with id " + messageId |
| + " on queue with id " + queueId); |
| } |
| |
| getLogger().debug("Dequeuing message {} on queue with id {}", messageId, queueId); |
| |
| } |
| |
| } |
| catch (SQLException e) |
| { |
| getLogger().error("Failed to dequeue message {}", messageId, e); |
| throw new StoreException("Error deleting enqueued message with id " + messageId + " for queue with id " |
| + queueId + " from database", e); |
| } |
| |
| } |
| |
| private void removeXid(ConnectionWrapper connWrapper, long format, byte[] globalId, byte[] branchId) |
| throws StoreException |
| { |
| Connection conn = connWrapper.getConnection(); |
| |
| |
| try |
| { |
| PreparedStatement stmt = conn.prepareStatement("DELETE FROM " + getXidTableName() |
| + " WHERE format = ? and global_id = ? and branch_id = ?"); |
| try |
| { |
| stmt.setLong(1,format); |
| stmt.setBytes(2,globalId); |
| stmt.setBytes(3,branchId); |
| int results = stmt.executeUpdate(); |
| |
| |
| |
| if(results != 1) |
| { |
| 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 |
| { |
| stmt.setLong(1,format); |
| stmt.setBytes(2,globalId); |
| stmt.setBytes(3,branchId); |
| int results = stmt.executeUpdate(); |
| |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| |
| } |
| catch (SQLException e) |
| { |
| getLogger().error("Failed to remove xid", e); |
| throw new StoreException("Error deleting enqueued message with xid", e); |
| } |
| |
| } |
| |
| private List<Runnable> recordXid(ConnectionWrapper connWrapper, long format, byte[] globalId, byte[] branchId, |
| Transaction.EnqueueRecord[] enqueues, Transaction.DequeueRecord[] dequeues) throws StoreException |
| { |
| Connection conn = connWrapper.getConnection(); |
| |
| |
| try |
| { |
| |
| PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getXidTableName() |
| + " ( format, global_id, branch_id ) values (?, ?, ?)"); |
| try |
| { |
| stmt.setLong(1,format); |
| stmt.setBytes(2, globalId); |
| stmt.setBytes(3, branchId); |
| stmt.executeUpdate(); |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| |
| for(Transaction.EnqueueRecord enqueue : enqueues) |
| { |
| StoredMessage storedMessage = enqueue.getMessage().getStoredMessage(); |
| if(storedMessage instanceof StoredJDBCMessage) |
| { |
| ((StoredJDBCMessage) storedMessage).store(conn); |
| } |
| } |
| |
| |
| stmt = conn.prepareStatement("INSERT INTO " + getXidActionsTableName() |
| + " ( format, global_id, branch_id, action_type, " + |
| "queue_id, message_id ) values (?,?,?,?,?,?) "); |
| |
| try |
| { |
| stmt.setLong(1,format); |
| stmt.setBytes(2, globalId); |
| stmt.setBytes(3, branchId); |
| |
| if(enqueues != null) |
| { |
| stmt.setString(4, "E"); |
| for(Transaction.EnqueueRecord record : enqueues) |
| { |
| stmt.setString(5, record.getResource().getId().toString()); |
| stmt.setLong(6, record.getMessage().getMessageNumber()); |
| stmt.executeUpdate(); |
| } |
| } |
| |
| if(dequeues != null) |
| { |
| stmt.setString(4, "D"); |
| for(Transaction.DequeueRecord record : dequeues) |
| { |
| stmt.setString(5, record.getEnqueueRecord().getQueueId().toString()); |
| stmt.setLong(6, record.getEnqueueRecord().getMessageNumber()); |
| stmt.executeUpdate(); |
| } |
| } |
| |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| return Collections.emptyList(); |
| } |
| catch (SQLException e) |
| { |
| getLogger().error("Failed to record xid", e); |
| throw new StoreException("Error writing xid ", e); |
| } |
| |
| } |
| |
| public void onOpen(final ConfiguredObject<?> parent) |
| { |
| } |
| |
| protected void setTablePrefix(final String tablePrefix) |
| { |
| _tablePrefix = tablePrefix; |
| } |
| |
| private String getDbVersionTableName() |
| { |
| return _tablePrefix + DB_VERSION_TABLE_NAME_SUFFIX; |
| } |
| |
| private String getQueueEntryTableName() |
| { |
| return _tablePrefix + QUEUE_ENTRY_TABLE_NAME_SUFFIX; |
| } |
| |
| private String getMetaDataTableName() |
| { |
| return _tablePrefix + META_DATA_TABLE_NAME_SUFFIX; |
| } |
| |
| private String getMessageContentTableName() |
| { |
| return _tablePrefix + MESSAGE_CONTENT_TABLE_NAME_SUFFIX; |
| } |
| |
| private String getXidTableName() |
| { |
| return _tablePrefix + XID_TABLE_NAME_SUFFIX; |
| } |
| |
| private String getXidActionsTableName() |
| { |
| return _tablePrefix + XID_ACTIONS_TABLE_NAME_SUFFIX; |
| } |
| |
| private static final class ConnectionWrapper |
| { |
| private final Connection _connection; |
| |
| public ConnectionWrapper(Connection conn) |
| { |
| _connection = conn; |
| } |
| |
| public Connection getConnection() |
| { |
| return _connection; |
| } |
| } |
| |
| |
| private void commitTran(ConnectionWrapper connWrapper) throws StoreException |
| { |
| try |
| { |
| Connection conn = connWrapper.getConnection(); |
| conn.commit(); |
| |
| getLogger().debug("commit tran completed"); |
| |
| conn.close(); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error commit tx", e); |
| } |
| } |
| |
| private <X> ListenableFuture<X> commitTranAsync(final ConnectionWrapper connWrapper, final X val) throws StoreException |
| { |
| final SettableFuture<X> future = SettableFuture.create(); |
| _executor.submit(new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| try |
| { |
| commitTran(connWrapper); |
| future.set(val); |
| } |
| catch (RuntimeException e) |
| { |
| future.setException(e); |
| } |
| } |
| }); |
| return future; |
| } |
| |
| private void abortTran(ConnectionWrapper connWrapper) throws StoreException |
| { |
| if (connWrapper == null) |
| { |
| throw new StoreException("Fatal internal error: transactional context is empty at abortTran"); |
| } |
| |
| getLogger().debug("abort tran called: {}", connWrapper.getConnection()); |
| |
| try |
| { |
| Connection conn = connWrapper.getConnection(); |
| conn.rollback(); |
| conn.close(); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error aborting transaction: " + e.getMessage(), e); |
| } |
| |
| } |
| |
| private void storeMetaData(Connection conn, long messageId, StorableMessageMetaData metaData) |
| throws SQLException |
| { |
| getLogger().debug("Adding metadata for message {}", messageId); |
| |
| PreparedStatement stmt = conn.prepareStatement("INSERT INTO " + getMetaDataTableName() |
| + "( message_id , meta_data ) values (?, ?)"); |
| try |
| { |
| stmt.setLong(1, messageId); |
| |
| final int bodySize = 1 + metaData.getStorableSize(); |
| byte[] underlying = new byte[bodySize]; |
| underlying[0] = (byte) metaData.getType().ordinal(); |
| QpidByteBuffer buf = QpidByteBuffer.wrap(underlying); |
| buf.position(1); |
| buf = buf.slice(); |
| |
| metaData.writeToBuffer(buf); |
| ByteArrayInputStream bis = new ByteArrayInputStream(underlying); |
| try |
| { |
| stmt.setBinaryStream(2, bis, underlying.length); |
| int result = stmt.executeUpdate(); |
| |
| if (result == 0) |
| { |
| throw new StoreException("Unable to add meta data for message " + messageId); |
| } |
| } |
| finally |
| { |
| try |
| { |
| bis.close(); |
| } |
| catch (IOException e) |
| { |
| |
| throw new SQLException(e); |
| } |
| } |
| |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| |
| } |
| |
| |
| private static class RecordImpl implements Transaction.EnqueueRecord, Transaction.DequeueRecord, TransactionLogResource, EnqueueableMessage |
| { |
| |
| private final JDBCEnqueueRecord _record; |
| private long _messageNumber; |
| private UUID _queueId; |
| |
| public RecordImpl(UUID queueId, long messageNumber) |
| { |
| _messageNumber = messageNumber; |
| _queueId = queueId; |
| _record = new JDBCEnqueueRecord(queueId, messageNumber); |
| } |
| |
| @Override |
| public MessageEnqueueRecord getEnqueueRecord() |
| { |
| return _record; |
| } |
| |
| @Override |
| public TransactionLogResource getResource() |
| { |
| return this; |
| } |
| |
| @Override |
| public EnqueueableMessage getMessage() |
| { |
| return this; |
| } |
| |
| @Override |
| public long getMessageNumber() |
| { |
| return _messageNumber; |
| } |
| |
| @Override |
| public boolean isPersistent() |
| { |
| return true; |
| } |
| |
| @Override |
| public StoredMessage getStoredMessage() |
| { |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public String getName() |
| { |
| return _queueId.toString(); |
| } |
| |
| @Override |
| public UUID getId() |
| { |
| return _queueId; |
| } |
| |
| @Override |
| public MessageDurability getMessageDurability() |
| { |
| return MessageDurability.DEFAULT; |
| } |
| } |
| |
| private StorableMessageMetaData getMetaData(long messageId) throws SQLException |
| { |
| |
| Connection conn = newAutoCommitConnection(); |
| try |
| { |
| PreparedStatement stmt = conn.prepareStatement("SELECT meta_data FROM " + getMetaDataTableName() |
| + " WHERE message_id = ?"); |
| try |
| { |
| stmt.setLong(1,messageId); |
| ResultSet rs = stmt.executeQuery(); |
| try |
| { |
| |
| if(rs.next()) |
| { |
| byte[] dataAsBytes = getBlobAsBytes(rs, 1); |
| return getStorableMessageMetaData(dataAsBytes); |
| } |
| else |
| { |
| 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; |
| |
| private void addContent(final Connection conn, long messageId, |
| Collection<QpidByteBuffer> contentBody) |
| { |
| getLogger().debug("Adding content for message {}", messageId); |
| |
| PreparedStatement stmt = null; |
| |
| int size = 0; |
| |
| for(QpidByteBuffer buf : contentBody) |
| { |
| size += buf.remaining(); |
| } |
| byte[] data = new byte[size]; |
| ByteBuffer dst = ByteBuffer.wrap(data); |
| for(QpidByteBuffer buf : contentBody) |
| { |
| buf.copyTo(dst); |
| } |
| |
| try |
| { |
| |
| stmt = conn.prepareStatement("INSERT INTO " + getMessageContentTableName() |
| + "( message_id, content ) values (?, ?)"); |
| stmt.setLong(1, messageId); |
| stmt.setBinaryStream(2, new ByteArrayInputStream(data), data.length); |
| stmt.executeUpdate(); |
| } |
| catch (SQLException e) |
| { |
| 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 |
| { |
| conn = newAutoCommitConnection(); |
| |
| stmt = conn.prepareStatement("SELECT content FROM " + getMessageContentTableName() |
| + " WHERE message_id = ?"); |
| stmt.setLong(1,messageId); |
| ResultSet rs = stmt.executeQuery(); |
| |
| if (rs.next()) |
| { |
| byte[] data = getBlobAsBytes(rs, 1); |
| int offset = 0; |
| int length = data.length; |
| Collection<QpidByteBuffer> buffers = QpidByteBuffer.allocateDirectCollection(length); |
| for(QpidByteBuffer buf : buffers) |
| { |
| int bufSize = buf.remaining(); |
| buf.put(data, offset, bufSize); |
| buf.flip(); |
| offset+=bufSize; |
| } |
| return buffers; |
| } |
| else |
| { |
| throw new StoreException("Unable to find message with id " + messageId); |
| } |
| |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error retrieving content for message " + messageId + ": " + e.getMessage(), e); |
| } |
| finally |
| { |
| JdbcUtils.closePreparedStatement(stmt, getLogger()); |
| JdbcUtils.closeConnection(conn, getLogger()); |
| } |
| } |
| |
| @Override |
| public boolean isPersistent() |
| { |
| return true; |
| } |
| |
| @Override |
| public long getBytesEvacuatedFromMemory() |
| { |
| return _bytesEvacuatedFromMemory.get(); |
| } |
| |
| protected class JDBCTransaction implements Transaction |
| { |
| private final ConnectionWrapper _connWrapper; |
| private int _storeSizeIncrease; |
| private final List<Runnable> _preCommitActions = new ArrayList<>(); |
| private final List<Runnable> _postCommitActions = new ArrayList<>(); |
| |
| protected JDBCTransaction() |
| { |
| try |
| { |
| _connWrapper = new ConnectionWrapper(newConnection()); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException(e); |
| } |
| } |
| |
| @Override |
| public MessageEnqueueRecord enqueueMessage(TransactionLogResource queue, EnqueueableMessage message) |
| { |
| checkMessageStoreOpen(); |
| |
| final StoredMessage storedMessage = message.getStoredMessage(); |
| if(storedMessage instanceof StoredJDBCMessage) |
| { |
| _preCommitActions.add(new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| try |
| { |
| ((StoredJDBCMessage) storedMessage).store(_connWrapper.getConnection()); |
| _storeSizeIncrease += storedMessage.getMetaData().getContentSize(); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Exception on enqueuing message into message store" + _messageId, |
| e); |
| } |
| } |
| }); |
| } |
| AbstractJDBCMessageStore.this.enqueueMessage(_connWrapper, queue, message.getMessageNumber()); |
| return new JDBCEnqueueRecord(queue.getId(), message.getMessageNumber()); |
| } |
| |
| @Override |
| public void dequeueMessage(final MessageEnqueueRecord enqueueRecord) |
| { |
| checkMessageStoreOpen(); |
| |
| AbstractJDBCMessageStore.this.dequeueMessage(_connWrapper, |
| enqueueRecord.getQueueId(), |
| enqueueRecord.getMessageNumber()); |
| } |
| |
| @Override |
| public void commitTran() |
| { |
| checkMessageStoreOpen(); |
| doPreCommitActions(); |
| AbstractJDBCMessageStore.this.commitTran(_connWrapper); |
| storedSizeChange(_storeSizeIncrease); |
| doPostCommitActions(); |
| } |
| |
| @Override |
| public <X> ListenableFuture<X> commitTranAsync(final X val) |
| { |
| checkMessageStoreOpen(); |
| doPreCommitActions(); |
| ListenableFuture<X> futureResult = AbstractJDBCMessageStore.this.commitTranAsync(_connWrapper, val); |
| storedSizeChange(_storeSizeIncrease); |
| doPostCommitActions(); |
| return futureResult; |
| } |
| |
| private void doPreCommitActions() |
| { |
| for(Runnable action : _preCommitActions) |
| { |
| action.run(); |
| } |
| _preCommitActions.clear(); |
| } |
| |
| private void doPostCommitActions() |
| { |
| // QPID-7447: prevent unnecessary allocation of empty iterator |
| if (!_postCommitActions.isEmpty()) |
| { |
| for (Runnable action : _postCommitActions) |
| { |
| action.run(); |
| } |
| _postCommitActions.clear(); |
| } |
| } |
| |
| @Override |
| public void abortTran() |
| { |
| checkMessageStoreOpen(); |
| _preCommitActions.clear(); |
| AbstractJDBCMessageStore.this.abortTran(_connWrapper); |
| } |
| |
| @Override |
| public void removeXid(final StoredXidRecord record) |
| { |
| checkMessageStoreOpen(); |
| |
| AbstractJDBCMessageStore.this.removeXid(_connWrapper, |
| record.getFormat(), |
| record.getGlobalId(), |
| record.getBranchId()); |
| } |
| |
| @Override |
| public StoredXidRecord recordXid(final long format, |
| final byte[] globalId, |
| final byte[] branchId, |
| EnqueueRecord[] enqueues, |
| DequeueRecord[] dequeues) |
| { |
| checkMessageStoreOpen(); |
| |
| _postCommitActions.addAll(AbstractJDBCMessageStore.this.recordXid(_connWrapper, format, globalId, branchId, enqueues, dequeues)); |
| return new JDBCStoredXidRecord(format, globalId, branchId); |
| } |
| |
| |
| } |
| |
| private static class JDBCStoredXidRecord implements Transaction.StoredXidRecord |
| { |
| private final long _format; |
| private final byte[] _globalId; |
| private final byte[] _branchId; |
| |
| public JDBCStoredXidRecord(final long format, final byte[] globalId, final byte[] branchId) |
| { |
| _format = format; |
| _globalId = globalId; |
| _branchId = branchId; |
| } |
| |
| @Override |
| public long getFormat() |
| { |
| return _format; |
| } |
| |
| @Override |
| public byte[] getGlobalId() |
| { |
| return _globalId; |
| } |
| |
| @Override |
| public byte[] getBranchId() |
| { |
| return _branchId; |
| } |
| |
| |
| @Override |
| public boolean equals(final Object o) |
| { |
| if (this == o) |
| { |
| return true; |
| } |
| if (o == null || getClass() != o.getClass()) |
| { |
| return false; |
| } |
| |
| final JDBCStoredXidRecord that = (JDBCStoredXidRecord) o; |
| |
| return _format == that._format |
| && Arrays.equals(_globalId, that._globalId) |
| && Arrays.equals(_branchId, that._branchId); |
| |
| } |
| |
| @Override |
| public int hashCode() |
| { |
| int result = (int) (_format ^ (_format >>> 32)); |
| result = 31 * result + Arrays.hashCode(_globalId); |
| result = 31 * result + Arrays.hashCode(_branchId); |
| return result; |
| } |
| } |
| |
| static interface MessageDataRef<T extends StorableMessageMetaData> |
| { |
| T getMetaData(); |
| Collection<QpidByteBuffer> getData(); |
| void setData(Collection<QpidByteBuffer> data); |
| boolean isHardRef(); |
| long clear(); |
| } |
| |
| private static final class MessageDataHardRef<T extends StorableMessageMetaData> implements MessageDataRef<T> |
| { |
| private final T _metaData; |
| private volatile Collection<QpidByteBuffer> _data; |
| |
| private MessageDataHardRef(final T metaData) |
| { |
| _metaData = metaData; |
| } |
| |
| @Override |
| public T getMetaData() |
| { |
| return _metaData; |
| } |
| |
| @Override |
| public Collection<QpidByteBuffer> getData() |
| { |
| return _data; |
| } |
| |
| @Override |
| public void setData(final Collection<QpidByteBuffer> data) |
| { |
| _data = data; |
| } |
| |
| @Override |
| public long clear() |
| { |
| long bytesCleared = 0; |
| if(_metaData != null) |
| { |
| bytesCleared += _metaData.getStorableSize(); |
| _metaData.clearEncodedForm(); |
| } |
| if(_data != null) |
| { |
| for(QpidByteBuffer buf : _data) |
| { |
| bytesCleared += buf.remaining(); |
| buf.dispose(); |
| } |
| _data = null; |
| } |
| return bytesCleared; |
| } |
| |
| @Override |
| public boolean isHardRef() |
| { |
| return true; |
| } |
| } |
| |
| private static final class MessageDataSoftRef<T extends StorableMessageMetaData> implements MessageDataRef<T> |
| { |
| |
| private T _metaData; |
| private volatile Collection<QpidByteBuffer> _data; |
| |
| private MessageDataSoftRef(final T metaData, Collection<QpidByteBuffer> data) |
| { |
| _metaData = metaData; |
| _data = data; |
| } |
| |
| @Override |
| public T getMetaData() |
| { |
| return _metaData; |
| } |
| |
| @Override |
| public Collection<QpidByteBuffer> getData() |
| { |
| return _data; |
| } |
| |
| @Override |
| public void setData(final Collection<QpidByteBuffer> data) |
| { |
| _data = data; |
| } |
| |
| @Override |
| public long clear() |
| { |
| long bytesCleared = 0; |
| if(_metaData != null) |
| { |
| bytesCleared += _metaData.getStorableSize(); |
| _metaData.clearEncodedForm(); |
| _metaData = null; |
| } |
| if(_data != null) |
| { |
| for(QpidByteBuffer buf : _data) |
| { |
| bytesCleared += buf.remaining(); |
| buf.dispose(); |
| } |
| _data = null; |
| } |
| return bytesCleared; |
| } |
| |
| @Override |
| public boolean isHardRef() |
| { |
| return false; |
| } |
| } |
| |
| private class StoredJDBCMessage<T extends StorableMessageMetaData> implements StoredMessage<T>, MessageHandle<T> |
| { |
| |
| private final long _messageId; |
| |
| private MessageDataRef<T> _messageDataRef; |
| |
| |
| StoredJDBCMessage(long messageId, T metaData) |
| { |
| this(messageId, metaData, false); |
| } |
| |
| |
| StoredJDBCMessage(long messageId, |
| T metaData, boolean isRecovered) |
| { |
| _messageId = messageId; |
| |
| if(!isRecovered) |
| { |
| _messageDataRef = new MessageDataHardRef<>(metaData); |
| } |
| else |
| { |
| _messageDataRef = new MessageDataSoftRef<>(metaData, null); |
| } |
| } |
| |
| |
| @Override |
| public synchronized T getMetaData() |
| { |
| if (_messageDataRef == null) |
| { |
| return null; |
| } |
| else |
| { |
| T metaData = _messageDataRef.getMetaData(); |
| |
| if (metaData == null) |
| { |
| checkMessageStoreOpen(); |
| try |
| { |
| metaData = (T) AbstractJDBCMessageStore.this.getMetaData(_messageId); |
| _messageDataRef = new MessageDataSoftRef<>(metaData, _messageDataRef.getData()); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Failed to get metadata for message id: " + _messageId, e); |
| } |
| } |
| return metaData; |
| } |
| } |
| |
| @Override |
| public long getMessageNumber() |
| { |
| return _messageId; |
| } |
| |
| @Override |
| public synchronized void addContent(QpidByteBuffer src) |
| { |
| src = src.slice(); |
| Collection<QpidByteBuffer> data = _messageDataRef.getData(); |
| if(data == null) |
| { |
| _messageDataRef.setData(Collections.singleton(src)); |
| } |
| else |
| { |
| List<QpidByteBuffer> newCollection = new ArrayList<>(data.size()+1); |
| newCollection.addAll(data); |
| newCollection.add(src); |
| _messageDataRef.setData(Collections.unmodifiableCollection(newCollection)); |
| } |
| |
| } |
| |
| @Override |
| public StoredMessage<T> allContentAdded() |
| { |
| return this; |
| } |
| |
| /** |
| * returns QBBs containing the content. The caller must not dispose of them because we keep a reference in _messageDataRef. |
| */ |
| private Collection<QpidByteBuffer> getContentAsByteBuffer() |
| { |
| Collection<QpidByteBuffer> data = _messageDataRef == null ? Collections.<QpidByteBuffer>emptyList() : _messageDataRef.getData(); |
| if(data == null) |
| { |
| if(stored()) |
| { |
| checkMessageStoreOpen(); |
| data = AbstractJDBCMessageStore.this.getAllContent(_messageId); |
| _messageDataRef.setData(data); |
| } |
| else |
| { |
| data = Collections.emptyList(); |
| } |
| } |
| return data; |
| } |
| |
| @Override |
| public synchronized Collection<QpidByteBuffer> getContent(int offset, int length) |
| { |
| Collection<QpidByteBuffer> bufs = getContentAsByteBuffer(); |
| Collection<QpidByteBuffer> content = new ArrayList<>(bufs.size()); |
| |
| int pos = 0; |
| for (QpidByteBuffer buf : bufs) |
| { |
| if (length > 0) |
| { |
| int bufRemaining = buf.remaining(); |
| if (pos + bufRemaining <= offset) |
| { |
| pos += bufRemaining; |
| } |
| else if (pos >= offset) |
| { |
| buf = buf.duplicate(); |
| if (bufRemaining <= length) |
| { |
| length -= bufRemaining; |
| } |
| else |
| { |
| buf.limit(length); |
| length = 0; |
| } |
| content.add(buf); |
| pos += buf.remaining(); |
| |
| } |
| else |
| { |
| int offsetInBuf = offset - pos; |
| int limit = length < bufRemaining - offsetInBuf ? length : bufRemaining - offsetInBuf; |
| final QpidByteBuffer bufView = buf.view(offsetInBuf, limit); |
| content.add(bufView); |
| length -= limit; |
| pos+=limit+offsetInBuf; |
| } |
| } |
| |
| } |
| return content; |
| } |
| |
| synchronized void store(final Connection conn) throws SQLException |
| { |
| if (!stored()) |
| { |
| AbstractJDBCMessageStore.this.storeMetaData(conn, _messageId, _messageDataRef.getMetaData()); |
| AbstractJDBCMessageStore.this.addContent(conn, _messageId, |
| _messageDataRef.getData() == null |
| ? Collections.<QpidByteBuffer>emptySet() |
| : _messageDataRef.getData()); |
| |
| getLogger().debug("Storing message {} to store", _messageId); |
| |
| MessageDataRef<T> hardRef = _messageDataRef; |
| MessageDataSoftRef<T> messageDataSoftRef; |
| |
| messageDataSoftRef = new MessageDataSoftRef<>(hardRef.getMetaData(), hardRef.getData()); |
| |
| _messageDataRef = messageDataSoftRef; |
| |
| |
| } |
| } |
| |
| synchronized ListenableFuture<Void> flushToStore() |
| { |
| if (_messageDataRef != null) |
| { |
| if(!stored()) |
| { |
| try (Connection conn = newConnection()) |
| { |
| store(conn); |
| conn.commit(); |
| storedSizeChange(getMetaData().getContentSize()); |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Failed to flow to disk", e); |
| } |
| } |
| |
| } |
| return Futures.immediateFuture(null); |
| } |
| |
| @Override |
| public synchronized void remove() |
| { |
| getLogger().debug("REMOVE called on message: {}", _messageId); |
| |
| checkMessageStoreOpen(); |
| Collection<QpidByteBuffer> data = _messageDataRef.getData(); |
| |
| final T metaData = getMetaData(); |
| int delta = metaData.getContentSize(); |
| _messages.remove(this); |
| if(stored()) |
| { |
| AbstractJDBCMessageStore.this.removeMessage(_messageId); |
| storedSizeChange(-delta); |
| } |
| if(data != null) |
| { |
| _messageDataRef.setData(null); |
| for(QpidByteBuffer buf : data) |
| { |
| buf.dispose(); |
| } |
| } |
| metaData.dispose(); |
| _messageDataRef = null; |
| } |
| |
| @Override |
| public synchronized boolean isInMemory() |
| { |
| return _messageDataRef != null && (_messageDataRef.isHardRef() || _messageDataRef.getData() != null); |
| } |
| |
| private boolean stored() |
| { |
| return _messageDataRef != null && !_messageDataRef.isHardRef(); |
| } |
| |
| @Override |
| public synchronized boolean flowToDisk() |
| { |
| |
| flushToStore(); |
| if(_messageDataRef != null && !_messageDataRef.isHardRef()) |
| { |
| final long bytesCleared = ((MessageDataSoftRef) _messageDataRef).clear(); |
| _bytesEvacuatedFromMemory.addAndGet(bytesCleared); |
| } |
| return true; |
| } |
| |
| public synchronized void clear() |
| { |
| if (_messageDataRef != null) |
| { |
| _messageDataRef.clear(); |
| } |
| } |
| |
| @Override |
| public String toString() |
| { |
| return this.getClass() + "[messageId=" + _messageId + "]"; |
| } |
| } |
| |
| @Override |
| public void addEventListener(EventListener eventListener, Event... events) |
| { |
| _eventManager.addEventListener(eventListener, events); |
| } |
| |
| @Override |
| public MessageStoreReader newMessageStoreReader() |
| { |
| return new JDBCMessageStoreReader(); |
| } |
| |
| private class JDBCMessageStoreReader implements MessageStoreReader |
| { |
| |
| @Override |
| public StoredMessage<?> getMessage(long messageId) throws StoreException |
| { |
| checkMessageStoreOpen(); |
| |
| Connection conn = null; |
| StoredJDBCMessage message; |
| try |
| { |
| conn = newAutoCommitConnection(); |
| try (PreparedStatement stmt = conn.prepareStatement("SELECT message_id, meta_data FROM " + getMetaDataTableName() |
| + " WHERE message_id = ?")) |
| { |
| stmt.setLong(1, messageId); |
| try (ResultSet rs = stmt.executeQuery()) |
| { |
| if (rs.next()) |
| { |
| byte[] dataAsBytes = getBlobAsBytes(rs, 2); |
| StorableMessageMetaData metaData = getStorableMessageMetaData(dataAsBytes); |
| message = createStoredJDBCMessage(messageId, metaData, true); |
| } |
| else |
| { |
| message = null; |
| } |
| } |
| } |
| return message; |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error encountered when visiting messages", e); |
| } |
| finally |
| { |
| JdbcUtils.closeConnection(conn, getLogger()); |
| } |
| } |
| |
| @Override |
| public void close() |
| { |
| |
| } |
| |
| |
| @Override |
| public void visitMessages(MessageHandler handler) throws StoreException |
| { |
| checkMessageStoreOpen(); |
| |
| Connection conn = null; |
| try |
| { |
| conn = newAutoCommitConnection(); |
| Statement stmt = conn.createStatement(); |
| try |
| { |
| ResultSet rs = stmt.executeQuery("SELECT message_id, meta_data FROM " + getMetaDataTableName()); |
| try |
| { |
| while (rs.next()) |
| { |
| long messageId = rs.getLong(1); |
| byte[] dataAsBytes = getBlobAsBytes(rs, 2); |
| QpidByteBuffer buf = QpidByteBuffer.wrap(dataAsBytes); |
| buf.position(1); |
| buf = buf.slice(); |
| MessageMetaDataType<?> type = MessageMetaDataTypeRegistry.fromOrdinal(((int)dataAsBytes[0]) &0xff); |
| StorableMessageMetaData metaData = type.createMetaData(buf); |
| buf.dispose(); |
| StoredJDBCMessage message = createStoredJDBCMessage(messageId, metaData, true); |
| if (!handler.handle(message)) |
| { |
| break; |
| } |
| } |
| } |
| finally |
| { |
| rs.close(); |
| } |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error encountered when visiting messages", e); |
| } |
| finally |
| { |
| JdbcUtils.closeConnection(conn, getLogger()); |
| } |
| } |
| |
| @Override |
| public void visitMessageInstances(TransactionLogResource queue, MessageInstanceHandler handler) |
| throws StoreException |
| { |
| checkMessageStoreOpen(); |
| |
| 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"); |
| try |
| { |
| stmt.setString(1, queue.getId().toString()); |
| ResultSet rs = stmt.executeQuery(); |
| try |
| { |
| while (rs.next()) |
| { |
| String id = rs.getString(1); |
| long messageId = rs.getLong(2); |
| UUID uuid = uuidFactory.createUuidFromString(id); |
| if (!handler.handle(new JDBCEnqueueRecord(uuid, messageId))) |
| { |
| break; |
| } |
| } |
| } |
| 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 |
| public void visitMessageInstances(MessageInstanceHandler handler) throws StoreException |
| { |
| checkMessageStoreOpen(); |
| |
| Connection conn = null; |
| try |
| { |
| CachingUUIDFactory uuidFactory = new CachingUUIDFactory(); |
| conn = newAutoCommitConnection(); |
| Statement stmt = conn.createStatement(); |
| try |
| { |
| ResultSet rs = stmt.executeQuery("SELECT queue_id, message_id FROM " + getQueueEntryTableName() |
| + " ORDER BY queue_id, message_id"); |
| try |
| { |
| while (rs.next()) |
| { |
| String id = rs.getString(1); |
| long messageId = rs.getLong(2); |
| UUID queueId = uuidFactory.createUuidFromString(id); |
| if (!handler.handle(new JDBCEnqueueRecord(queueId, messageId))) |
| { |
| break; |
| } |
| } |
| } |
| 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 |
| public void visitDistributedTransactions(DistributedTransactionHandler handler) throws StoreException |
| { |
| checkMessageStoreOpen(); |
| |
| Connection conn = null; |
| try |
| { |
| conn = newAutoCommitConnection(); |
| List<Xid> xids = new ArrayList<Xid>(); |
| |
| Statement stmt = conn.createStatement(); |
| try |
| { |
| ResultSet rs = stmt.executeQuery("SELECT format, global_id, branch_id FROM " + getXidTableName()); |
| try |
| { |
| while (rs.next()) |
| { |
| |
| long format = rs.getLong(1); |
| byte[] globalId = rs.getBytes(2); |
| byte[] branchId = rs.getBytes(3); |
| xids.add(new Xid(format, globalId, branchId)); |
| } |
| } |
| finally |
| { |
| rs.close(); |
| } |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| |
| |
| for (Xid xid : xids) |
| { |
| CachingUUIDFactory uuidFactory = new CachingUUIDFactory(); |
| 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 |
| { |
| pstmt.setLong(1, xid.getFormat()); |
| pstmt.setBytes(2, xid.getGlobalId()); |
| pstmt.setBytes(3, xid.getBranchId()); |
| |
| ResultSet rs = pstmt.executeQuery(); |
| try |
| { |
| while (rs.next()) |
| { |
| |
| String actionType = rs.getString(1); |
| UUID queueId = uuidFactory.createUuidFromString(rs.getString(2)); |
| long messageId = rs.getLong(3); |
| |
| RecordImpl record = new RecordImpl(queueId, messageId); |
| List<RecordImpl> records = "E".equals(actionType) ? enqueues : dequeues; |
| records.add(record); |
| } |
| } |
| finally |
| { |
| rs.close(); |
| } |
| } |
| finally |
| { |
| pstmt.close(); |
| } |
| |
| if (!handler.handle(new JDBCStoredXidRecord(xid.getFormat(), xid.getGlobalId(), xid.getBranchId()), |
| enqueues.toArray(new RecordImpl[enqueues.size()]), |
| dequeues.toArray(new RecordImpl[dequeues.size()]))) |
| { |
| break; |
| } |
| } |
| |
| } |
| catch (SQLException e) |
| { |
| throw new StoreException("Error encountered when visiting distributed transactions", e); |
| |
| } |
| finally |
| { |
| JdbcUtils.closeConnection(conn, getLogger()); |
| } |
| } |
| } |
| |
| private StorableMessageMetaData getStorableMessageMetaData(final byte[] dataAsBytes) |
| { |
| final QpidByteBuffer buf; |
| if (dataAsBytes.length > QpidByteBuffer.getPooledBufferSize()) |
| { |
| buf = QpidByteBuffer.wrap(dataAsBytes); |
| } |
| else |
| { |
| buf = QpidByteBuffer.allocateDirect(dataAsBytes.length); |
| buf.put(dataAsBytes); |
| buf.flip(); |
| } |
| buf.position(1); |
| QpidByteBuffer slice = buf.slice(); |
| int typeOrdinal = dataAsBytes[0] & 0xff; |
| MessageMetaDataType type = MessageMetaDataTypeRegistry.fromOrdinal(typeOrdinal); |
| StorableMessageMetaData metaData = type.createMetaData(slice); |
| slice.dispose(); |
| buf.dispose(); |
| return metaData; |
| } |
| |
| protected abstract void storedSizeChange(int storeSizeIncrease); |
| |
| @Override |
| public void onDelete(ConfiguredObject<?> parent) |
| { |
| // TODO should probably check we are closed |
| try |
| { |
| Connection conn = newAutoCommitConnection(); |
| try |
| { |
| |
| for (String tableName : getTableNames()) |
| { |
| Statement stmt = conn.createStatement(); |
| try |
| { |
| stmt.execute("DROP TABLE " + tableName); |
| } |
| catch(SQLException e) |
| { |
| getLogger().warn("Failed to drop table '{}'", tableName, e); |
| } |
| finally |
| { |
| stmt.close(); |
| } |
| } |
| } |
| finally |
| { |
| conn.close(); |
| } |
| } |
| catch(SQLException e) |
| { |
| getLogger().error("Exception while deleting store tables", e); |
| } |
| } |
| |
| public List<String> getTableNames() |
| { |
| return Arrays.asList(getDbVersionTableName(), |
| getMetaDataTableName(), |
| getMessageContentTableName(), |
| getQueueEntryTableName(), |
| getXidTableName(), |
| getXidActionsTableName()); |
| } |
| |
| |
| private static class JDBCEnqueueRecord implements MessageEnqueueRecord |
| { |
| private final UUID _queueId; |
| private final long _messageNumber; |
| |
| public JDBCEnqueueRecord(final UUID queueId, |
| final long messageNumber) |
| { |
| _queueId = queueId; |
| _messageNumber = messageNumber; |
| } |
| |
| public UUID getQueueId() |
| { |
| return _queueId; |
| } |
| |
| public long getMessageNumber() |
| { |
| return _messageNumber; |
| } |
| } |
| } |
