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 2012/03/27 13:04:05 UTC
svn commit: r1305809 [2/3] - in /qpid/trunk/qpid/java: ./ bdbstore/
bdbstore/bin/ bdbstore/src/main/java/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/entry/
bdbstore/src...
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/XidBinding.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/XidTB.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/XidBinding.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/XidBinding.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/XidTB.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/XidTB.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/XidBinding.java Tue Mar 27 11:04:02 2012
@@ -19,15 +19,27 @@
*
*/
-package org.apache.qpid.server.store.berkeleydb.tuples;
+package org.apache.qpid.server.store.berkeleydb.tuple;
import com.sleepycat.bind.tuple.TupleBinding;
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
-import org.apache.qpid.server.store.berkeleydb.keys.Xid;
-public class XidTB extends TupleBinding<Xid>
+import org.apache.qpid.server.store.berkeleydb.entry.Xid;
+
+public class XidBinding extends TupleBinding<Xid>
{
+
+ private static final XidBinding INSTANCE = new XidBinding();
+
+ public static XidBinding getInstance()
+ {
+ return INSTANCE;
+ }
+
+ /** private constructor forces getInstance instead */
+ private XidBinding() { }
+
@Override
public Xid entryToObject(TupleInput input)
{
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractStoreUpgrade.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractStoreUpgrade.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractStoreUpgrade.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/AbstractStoreUpgrade.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,83 @@
+/*
+ *
+ * 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.berkeleydb.upgrade;
+
+import java.util.List;
+
+import org.apache.log4j.Logger;
+
+import com.sleepycat.je.Database;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.Transaction;
+
+public abstract class AbstractStoreUpgrade implements StoreUpgrade
+{
+ private static final Logger _logger = Logger.getLogger(AbstractStoreUpgrade.class);
+ protected static final String[] USER_FRIENDLY_NAMES = new String[] { "Exchanges", "Queues", "Queue bindings",
+ "Message deliveries", "Message metadata", "Message content", "Bridges", "Links", "Distributed transactions" };
+
+ protected void reportFinished(Environment environment, String[] databaseNames, String[] userFriendlyNames)
+ {
+ if (_logger.isInfoEnabled())
+ {
+ _logger.info("Upgraded:");
+ List<String> databases = environment.getDatabaseNames();
+ for (int i = 0; i < databaseNames.length; i++)
+ {
+ if (databases.contains(databaseNames[i]))
+ {
+ _logger.info(" " + getRowCount(databaseNames[i], environment) + " rows in " + userFriendlyNames[i]);
+ }
+ }
+ }
+ }
+
+
+ protected void reportStarting(Environment environment, String[] databaseNames, String[] userFriendlyNames)
+ {
+ if (_logger.isInfoEnabled())
+ {
+ _logger.info("Upgrading:");
+ List<String> databases = environment.getDatabaseNames();
+ for (int i = 0; i < databaseNames.length; i++)
+ {
+ if (databases.contains(databaseNames[i]))
+ {
+ _logger.info(" " + getRowCount(databaseNames[i], environment) + " rows from " + userFriendlyNames[i]);
+ }
+ }
+ }
+ }
+
+ private long getRowCount(String databaseName, Environment environment)
+ {
+ DatabaseCallable<Long> operation = new DatabaseCallable<Long>()
+ {
+ @Override
+ public Long call(Database sourceDatabase, Database targetDatabase, Transaction transaction)
+ {
+ return sourceDatabase.count();
+ }
+ };
+ return new DatabaseTemplate(environment, databaseName, null).call(operation);
+ }
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorOperation.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorOperation.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorOperation.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorOperation.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,89 @@
+/*
+ *
+ * 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.berkeleydb.upgrade;
+
+import org.apache.log4j.Logger;
+
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.Transaction;
+
+public abstract class CursorOperation implements DatabaseRunnable
+{
+ private static final Logger _logger = Logger.getLogger(CursorOperation.class);
+
+ private CursorTemplate _template;
+ private long _rowCount;
+ private long _processedRowCount;
+
+ @Override
+ public void run(final Database sourceDatabase, final Database targetDatabase, final Transaction transaction)
+ {
+ _rowCount = sourceDatabase.count();
+ _template = new CursorTemplate(sourceDatabase, transaction, new DatabaseEntryCallback()
+ {
+ @Override
+ public void processEntry(final Database database, final Transaction transaction, final DatabaseEntry key,
+ final DatabaseEntry value)
+ {
+ _processedRowCount++;
+ CursorOperation.this.processEntry(database, targetDatabase, transaction, key, value);
+ if (getProcessedCount() % 1000 == 0)
+ {
+ _logger.info("Processed " + getProcessedCount() + " messages of " + getRowCount() + ".");
+ }
+ }
+
+ });
+ _template.processEntries();
+ }
+
+ public void abort()
+ {
+ if (_template != null)
+ {
+ _template.abort();
+ }
+ }
+
+ public boolean deleteCurrent()
+ {
+ if (_template != null)
+ {
+ return _template.deleteCurrent();
+ }
+ return false;
+ }
+
+ public long getRowCount()
+ {
+ return _rowCount;
+ }
+
+ public long getProcessedCount()
+ {
+ return _processedRowCount;
+ }
+
+ public abstract void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value);
+
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorTemplate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorTemplate.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorTemplate.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/CursorTemplate.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,75 @@
+package org.apache.qpid.server.store.berkeleydb.upgrade;
+
+/*
+*
+* 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.
+*
+*/
+import com.sleepycat.je.Cursor;
+import com.sleepycat.je.CursorConfig;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.LockMode;
+import com.sleepycat.je.OperationStatus;
+import com.sleepycat.je.Transaction;
+
+public class CursorTemplate
+{
+ private Database _database;
+ private Transaction _transaction;
+ private DatabaseEntryCallback _databaseEntryCallback;
+ private Cursor _cursor;
+ private boolean _iterating;
+
+ public CursorTemplate(Database database, Transaction transaction, DatabaseEntryCallback databaseEntryCallback)
+ {
+ _database = database;
+ _transaction = transaction;
+ _databaseEntryCallback = databaseEntryCallback;
+ }
+
+ public void processEntries()
+ {
+ _cursor = _database.openCursor(_transaction, CursorConfig.READ_COMMITTED);
+ DatabaseEntry key = new DatabaseEntry();
+ DatabaseEntry value = new DatabaseEntry();
+
+ try
+ {
+ _iterating = true;
+ while (_iterating && _cursor.getNext(key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS)
+ {
+ _databaseEntryCallback.processEntry(_database, _transaction, key, value);
+ }
+ }
+ finally
+ {
+ _cursor.close();
+ }
+ }
+
+ public boolean deleteCurrent()
+ {
+ return _cursor.delete() == OperationStatus.SUCCESS;
+ }
+
+ public void abort()
+ {
+ _iterating = false;
+ }
+}
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseCallable.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseCallable.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseCallable.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/QueueTuple.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseCallable.java Tue Mar 27 11:04:02 2012
@@ -18,8 +18,12 @@
* under the License.
*
*/
-package org.apache.qpid.server.store.berkeleydb.tuples;
+package org.apache.qpid.server.store.berkeleydb.upgrade;
-public interface QueueTuple
+import com.sleepycat.je.Database;
+import com.sleepycat.je.Transaction;
+
+public interface DatabaseCallable<T>
{
+ public T call(Database sourceDatabase, Database targetDatabase, Transaction transaction);
}
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseEntryCallback.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseEntryCallback.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseEntryCallback.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/MessageContentKey.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseEntryCallback.java Tue Mar 27 11:04:02 2012
@@ -18,25 +18,13 @@
* under the License.
*
*/
-package org.apache.qpid.server.store.berkeleydb;
+package org.apache.qpid.server.store.berkeleydb.upgrade;
-public class MessageContentKey
-{
- private long _messageId;
-
- public MessageContentKey(long messageId)
- {
- _messageId = messageId;
- }
-
-
- public long getMessageId()
- {
- return _messageId;
- }
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.Transaction;
- public void setMessageId(long messageId)
- {
- this._messageId = messageId;
- }
-}
\ No newline at end of file
+public interface DatabaseEntryCallback
+{
+ void processEntry(Database database, Transaction transaction, DatabaseEntry key, DatabaseEntry value);
+}
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseRunnable.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseRunnable.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseRunnable.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseRunnable.java Tue Mar 27 11:04:02 2012
@@ -18,23 +18,13 @@
* under the License.
*
*/
-package org.apache.qpid.server.store.berkeleydb.tuples;
+package org.apache.qpid.server.store.berkeleydb.upgrade;
-import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.Transaction;
-public abstract class TupleBindingFactory<E>
+public interface DatabaseRunnable
{
- private final int _version;
+ public void run(Database sourceDatabase, Database targetDatabase, Transaction transaction);
- public TupleBindingFactory(int version)
- {
- _version = version;
- }
-
- public abstract TupleBinding<E> getInstance();
-
- public int getVersion()
- {
- return _version;
- }
}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseTemplate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseTemplate.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseTemplate.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/DatabaseTemplate.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,114 @@
+/*
+ *
+ * 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.berkeleydb.upgrade;
+
+import org.apache.log4j.Logger;
+
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseConfig;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.Transaction;
+
+public class DatabaseTemplate
+{
+ private static final Logger _logger = Logger.getLogger(DatabaseTemplate.class);
+
+ private Environment _environment;
+ private String _sourceDatabaseName;
+ private String _targetDatabaseName;
+ private Transaction _parentTransaction;
+
+ public DatabaseTemplate(Environment environment, String sourceDatabaseName, Transaction transaction)
+ {
+ this(environment, sourceDatabaseName, null, transaction);
+ }
+
+ public DatabaseTemplate(Environment environment, String sourceDatabaseName, String targetDatabaseName,
+ Transaction parentTransaction)
+ {
+ _environment = environment;
+ _sourceDatabaseName = sourceDatabaseName;
+ _targetDatabaseName = targetDatabaseName;
+ _parentTransaction = parentTransaction;
+ }
+
+ public void run(DatabaseRunnable databaseRunnable)
+ {
+ DatabaseCallable<Void> callable = runnableToCallable(databaseRunnable);
+ call(callable);
+ }
+
+ public <T> T call(DatabaseCallable<T> databaseCallable)
+ {
+ Database sourceDatabase = null;
+ Database targetDatabase = null;
+ try
+ {
+ DatabaseConfig dbConfig = new DatabaseConfig();
+ dbConfig.setTransactional(true);
+ dbConfig.setAllowCreate(true);
+
+ sourceDatabase = _environment.openDatabase(_parentTransaction, _sourceDatabaseName, dbConfig);
+
+ if (_targetDatabaseName != null)
+ {
+ targetDatabase = _environment.openDatabase(_parentTransaction, _targetDatabaseName, dbConfig);
+ }
+
+ return databaseCallable.call(sourceDatabase, targetDatabase, _parentTransaction);
+ }
+ finally
+ {
+ closeDatabase(sourceDatabase);
+ closeDatabase(targetDatabase);
+ }
+ }
+
+ private DatabaseCallable<Void> runnableToCallable(final DatabaseRunnable databaseRunnable)
+ {
+ return new DatabaseCallable<Void>()
+ {
+
+ @Override
+ public Void call(Database sourceDatabase, Database targetDatabase, Transaction transaction)
+ {
+ databaseRunnable.run(sourceDatabase, targetDatabase, transaction);
+ return null;
+ }
+ };
+ }
+
+ private void closeDatabase(Database database)
+ {
+ if (database != null)
+ {
+ try
+ {
+ database.close();
+ }
+ catch (Exception e)
+ {
+ _logger.error("Unable to close database", e);
+ }
+ }
+ }
+
+}
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/TupleBindingFactory.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/StoreUpgrade.java Tue Mar 27 11:04:02 2012
@@ -18,23 +18,15 @@
* under the License.
*
*/
-package org.apache.qpid.server.store.berkeleydb.tuples;
+package org.apache.qpid.server.store.berkeleydb.upgrade;
-import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Environment;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.logging.LogSubject;
-public abstract class TupleBindingFactory<E>
+public interface StoreUpgrade
{
- private final int _version;
-
- public TupleBindingFactory(int version)
- {
- _version = version;
- }
-
- public abstract TupleBinding<E> getInstance();
-
- public int getVersion()
- {
- return _version;
- }
+ void performUpgrade(LogSubject logSubject, Environment environment, UpgradeInteractionHandler handler)
+ throws DatabaseException, AMQStoreException;
}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom4To5.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,926 @@
+/*
+ *
+ * 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.berkeleydb.upgrade;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.common.AMQPFilterTypes;
+import org.apache.qpid.exchange.ExchangeDefaults;
+import org.apache.qpid.framing.AMQFrameDecodingException;
+import org.apache.qpid.framing.AMQProtocolVersionException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.message.MessageMetaData;
+import org.apache.qpid.server.store.StorableMessageMetaData;
+import org.apache.qpid.server.store.berkeleydb.AMQShortStringEncoding;
+import org.apache.qpid.server.store.berkeleydb.FieldTableEncoding;
+
+import com.sleepycat.bind.tuple.ByteBinding;
+import com.sleepycat.bind.tuple.LongBinding;
+import com.sleepycat.bind.tuple.TupleBase;
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.Transaction;
+
+public class UpgradeFrom4To5 extends AbstractStoreUpgrade
+{
+ private static final String OLD_DELIVERY_DB = "deliveryDb_v4";
+ private static final String NEW_DELIVERY_DB = "deliveryDb_v5";
+ private static final String EXCHANGE_DB_NAME = "exchangeDb_v4";
+ private static final String OLD_BINDINGS_DB_NAME = "queueBindingsDb_v4";
+ private static final String NEW_BINDINGS_DB_NAME = "queueBindingsDb_v5";
+ private static final String OLD_QUEUE_DB_NAME = "queueDb_v4";
+ private static final String NEW_QUEUE_DB_NAME = "queueDb_v5";
+ private static final String OLD_METADATA_DB_NAME = "messageMetaDataDb_v4";
+ private static final String NEW_METADATA_DB_NAME = "messageMetaDataDb_v5";
+ private static final String OLD_CONTENT_DB_NAME = "messageContentDb_v4";
+ private static final String NEW_CONTENT_DB_NAME = "messageContentDb_v5";
+
+ private static final String[] OLD_DATABASE_NAMES = new String[] { EXCHANGE_DB_NAME, OLD_QUEUE_DB_NAME,
+ OLD_BINDINGS_DB_NAME, OLD_DELIVERY_DB, OLD_METADATA_DB_NAME, OLD_CONTENT_DB_NAME, "bridges_v4", "links_v4",
+ "xids_v4" };
+ private static final String[] NEW_DATABASE_NAMES = new String[] { "exchangeDb_v5", NEW_QUEUE_DB_NAME,
+ NEW_BINDINGS_DB_NAME, NEW_DELIVERY_DB, NEW_METADATA_DB_NAME, NEW_CONTENT_DB_NAME, "bridges_v5", "links_v5",
+ "xids_v5" };
+
+ private static final byte COLON = (byte) ':';
+
+ private static final Logger _logger = Logger.getLogger(UpgradeFrom4To5.class);
+
+ public void performUpgrade(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler) throws DatabaseException, AMQStoreException
+ {
+ _logger.info("Starting store upgrade from version 4");
+ Transaction transaction = null;
+ try
+ {
+ reportStarting(environment, OLD_DATABASE_NAMES, USER_FRIENDLY_NAMES);
+
+ transaction = environment.beginTransaction(null, null);
+
+ // find all queues which are bound to a topic exchange and which have a colon in their name
+ final List<AMQShortString> potentialDurableSubs = findPotentialDurableSubscriptions(logSubject, environment,
+ transaction);
+
+ Set<String> existingQueues = upgradeQueues(logSubject, environment, handler, potentialDurableSubs, transaction);
+ upgradeQueueBindings(logSubject, environment, handler, potentialDurableSubs, transaction);
+ Set<Long> messagesToDiscard = upgradeDelivery(logSubject, environment, existingQueues, handler, transaction);
+ upgradeContent(logSubject, environment, handler, messagesToDiscard, transaction);
+ upgradeMetaData(logSubject, environment, handler, messagesToDiscard, transaction);
+ renameRemainingDatabases(logSubject, environment, handler, transaction);
+ transaction.commit();
+
+ reportFinished(environment, NEW_DATABASE_NAMES, USER_FRIENDLY_NAMES);
+
+ }
+ catch (Exception e)
+ {
+ transaction.abort();
+ if (e instanceof DatabaseException)
+ {
+ throw (DatabaseException) e;
+ }
+ else if (e instanceof AMQStoreException)
+ {
+ throw (AMQStoreException) e;
+ }
+ else
+ {
+ throw new AMQStoreException("Unexpected exception", e);
+ }
+ }
+ }
+
+ private void upgradeQueueBindings(LogSubject logSubject, Environment environment, UpgradeInteractionHandler handler,
+ final List<AMQShortString> potentialDurableSubs, Transaction transaction)
+ {
+ if (environment.getDatabaseNames().contains(OLD_BINDINGS_DB_NAME))
+ {
+ _logger.info("Queue Bindings");
+ final BindingTuple bindingTuple = new BindingTuple();
+ CursorOperation databaseOperation = new CursorOperation()
+ {
+
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ // All the information required in binding entries is actually in the *key* not value.
+ BindingRecord oldBindingRecord = bindingTuple.entryToObject(key);
+
+ AMQShortString queueName = oldBindingRecord.getQueueName();
+ AMQShortString exchangeName = oldBindingRecord.getExchangeName();
+ AMQShortString routingKey = oldBindingRecord.getRoutingKey();
+ FieldTable arguments = oldBindingRecord.getArguments();
+
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug(String.format(
+ "Processing binding for queue %s, exchange %s, routingKey %s arguments %s", queueName,
+ exchangeName, routingKey, arguments));
+ }
+
+ // if the queue name is in the gathered list then inspect its binding arguments
+ // only topic exchange should have a JMS selector key in binding
+ if (potentialDurableSubs.contains(queueName)
+ && exchangeName.equals(ExchangeDefaults.TOPIC_EXCHANGE_NAME))
+ {
+ if (arguments == null)
+ {
+ arguments = new FieldTable();
+ }
+
+ AMQShortString selectorFilterKey = AMQPFilterTypes.JMS_SELECTOR.getValue();
+ if (!arguments.containsKey(selectorFilterKey))
+ {
+ if (_logger.isDebugEnabled())
+ {
+ _logger.info("adding the empty string (i.e. 'no selector') value for " + queueName
+ + " and exchange " + exchangeName);
+ }
+ arguments.put(selectorFilterKey, "");
+ }
+ }
+ addBindingToDatabase(bindingTuple, targetDatabase, transaction, queueName, exchangeName, routingKey,
+ arguments);
+ }
+ };
+ new DatabaseTemplate(environment, OLD_BINDINGS_DB_NAME, NEW_BINDINGS_DB_NAME, transaction)
+ .run(databaseOperation);
+ environment.removeDatabase(transaction, OLD_BINDINGS_DB_NAME);
+ _logger.info(databaseOperation.getRowCount() + " Queue Binding entries");
+ }
+ }
+
+ private Set<String> upgradeQueues(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler, List<AMQShortString> potentialDurableSubs, Transaction transaction)
+ {
+ _logger.info("Queues");
+ final Set<String> existingQueues = new HashSet<String>();
+ if (environment.getDatabaseNames().contains(OLD_QUEUE_DB_NAME))
+ {
+ final QueueRecordBinding binding = new QueueRecordBinding(potentialDurableSubs);
+ CursorOperation databaseOperation = new CursorOperation()
+ {
+ @Override
+ public void processEntry(final Database sourceDatabase, final Database targetDatabase,
+ final Transaction transaction, final DatabaseEntry key, final DatabaseEntry value)
+ {
+ QueueRecord record = binding.entryToObject(value);
+ DatabaseEntry newValue = new DatabaseEntry();
+ binding.objectToEntry(record, newValue);
+ targetDatabase.put(transaction, key, newValue);
+ existingQueues.add(record.getNameShortString().asString());
+ sourceDatabase.delete(transaction, key);
+ }
+ };
+ new DatabaseTemplate(environment, OLD_QUEUE_DB_NAME, NEW_QUEUE_DB_NAME, transaction).run(databaseOperation);
+ environment.removeDatabase(transaction, OLD_QUEUE_DB_NAME);
+ _logger.info(databaseOperation.getRowCount() + " Queue entries");
+ }
+ return existingQueues;
+ }
+
+ private List<AMQShortString> findPotentialDurableSubscriptions(final LogSubject logSubject,
+ final Environment environment, Transaction transaction)
+ {
+ final List<AMQShortString> exchangeNames = findTopicExchanges(logSubject, environment);
+ final List<AMQShortString> queues = new ArrayList<AMQShortString>();
+ final PartialBindingRecordBinding binding = new PartialBindingRecordBinding();
+
+ CursorOperation databaseOperation = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ PartialBindingRecord record = binding.entryToObject(key);
+ if (exchangeNames.contains(record.getExchangeName()) && record.getQueueName().contains(COLON))
+ {
+ queues.add(record.getQueueName());
+ }
+ }
+ };
+ new DatabaseTemplate(environment, OLD_BINDINGS_DB_NAME, transaction).run(databaseOperation);
+ return queues;
+ }
+
+ private Set<Long> upgradeDelivery(final LogSubject logSubject, final Environment environment,
+ final Set<String> existingQueues, final UpgradeInteractionHandler handler, Transaction transaction)
+ {
+ final Set<Long> messagesToDiscard = new HashSet<Long>();
+ final Set<String> queuesToDiscard = new HashSet<String>();
+ final QueueEntryKeyBinding queueEntryKeyBinding = new QueueEntryKeyBinding();
+ _logger.info("Delivery Records");
+
+ CursorOperation databaseOperation = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ QueueEntryKey entryKey = queueEntryKeyBinding.entryToObject(key);
+ Long messageId = entryKey.getMessageId();
+ final String queueName = entryKey.getQueueName().asString();
+ if (!existingQueues.contains(queueName))
+ {
+ if (queuesToDiscard.contains(queueName))
+ {
+ messagesToDiscard.add(messageId);
+ }
+ else
+ {
+ String lineSeparator = System.getProperty("line.separator");
+ String question = MessageFormat.format("Found persistent messages for non-durable queue ''{1}''. "
+ + " Do you with to create this queue and move all the messages into it?" + lineSeparator
+ + "NOTE: Answering No will result in these messages being discarded!", queueName);
+ UpgradeInteractionResponse response = handler.requireResponse(question.toString(),
+ UpgradeInteractionResponse.YES, UpgradeInteractionResponse.YES,
+ UpgradeInteractionResponse.NO, UpgradeInteractionResponse.ABORT);
+
+ if (response == UpgradeInteractionResponse.YES)
+ {
+ createQueue(environment, transaction, queueName);
+ existingQueues.add(queueName);
+ }
+ else if (response == UpgradeInteractionResponse.NO)
+ {
+ queuesToDiscard.add(queueName);
+ messagesToDiscard.add(messageId);
+ }
+ else
+ {
+ throw new RuntimeException("Unable is aborted!");
+ }
+ }
+ }
+
+ if (!messagesToDiscard.contains(messageId))
+ {
+ DatabaseEntry newKey = new DatabaseEntry();
+ queueEntryKeyBinding.objectToEntry(entryKey, newKey);
+ targetDatabase.put(transaction, newKey, value);
+
+ }
+ }
+ };
+ new DatabaseTemplate(environment, OLD_DELIVERY_DB, NEW_DELIVERY_DB, transaction).run(databaseOperation);
+
+ if (!messagesToDiscard.isEmpty())
+ {
+ databaseOperation = new CursorOperation()
+ {
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ QueueEntryKey entryKey = queueEntryKeyBinding.entryToObject(key);
+ Long messageId = entryKey.getMessageId();
+
+ if (messagesToDiscard.contains(messageId))
+ {
+ messagesToDiscard.remove(messageId);
+ }
+ }
+ };
+ new DatabaseTemplate(environment, NEW_DELIVERY_DB, transaction).run(databaseOperation);
+ }
+ _logger.info(databaseOperation.getRowCount() + " Delivery Records entries ");
+ environment.removeDatabase(transaction, OLD_DELIVERY_DB);
+
+ return messagesToDiscard;
+ }
+
+ protected void createQueue(final Environment environment, Transaction transaction, final String queueName)
+ {
+
+ final QueueRecordBinding binding = new QueueRecordBinding(null);
+ final BindingTuple bindingTuple = new BindingTuple();
+ DatabaseRunnable queueCreateOperation = new DatabaseRunnable()
+ {
+
+ @Override
+ public void run(Database newQueueDatabase, Database newBindingsDatabase, Transaction transaction)
+ {
+ AMQShortString queueNameAMQ = new AMQShortString(queueName);
+ QueueRecord record = new QueueRecord(queueNameAMQ, null, false, null);
+
+ DatabaseEntry key = new DatabaseEntry();
+
+ TupleOutput output = new TupleOutput();
+ AMQShortStringEncoding.writeShortString(record.getNameShortString(), output);
+ TupleBase.outputToEntry(output, key);
+
+ DatabaseEntry newValue = new DatabaseEntry();
+ binding.objectToEntry(record, newValue);
+ newQueueDatabase.put(transaction, key, newValue);
+
+ FieldTable emptyArguments = new FieldTable();
+ addBindingToDatabase(bindingTuple, newBindingsDatabase, transaction, queueNameAMQ,
+ ExchangeDefaults.DIRECT_EXCHANGE_NAME, queueNameAMQ, emptyArguments);
+
+ // TODO QPID-3490 we should not persist a default exchange binding
+ addBindingToDatabase(bindingTuple, newBindingsDatabase, transaction, queueNameAMQ,
+ ExchangeDefaults.DEFAULT_EXCHANGE_NAME, queueNameAMQ, emptyArguments);
+ }
+ };
+ new DatabaseTemplate(environment, NEW_QUEUE_DB_NAME, NEW_BINDINGS_DB_NAME, transaction).run(queueCreateOperation);
+ }
+
+ private List<AMQShortString> findTopicExchanges(final LogSubject logSubject, final Environment environment)
+ {
+ final List<AMQShortString> topicExchanges = new ArrayList<AMQShortString>();
+ final ExchangeRecordBinding binding = new ExchangeRecordBinding();
+ CursorOperation databaseOperation = new CursorOperation()
+ {
+
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ ExchangeRecord record = binding.entryToObject(value);
+ if (ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(record.getType()))
+ {
+ topicExchanges.add(record.getName());
+ }
+ }
+ };
+ new DatabaseTemplate(environment, EXCHANGE_DB_NAME, null).run(databaseOperation);
+ return topicExchanges;
+ }
+
+ private void upgradeMetaData(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler, final Set<Long> messagesToDiscard, Transaction transaction)
+ {
+ _logger.info("Message MetaData");
+ if (environment.getDatabaseNames().contains(OLD_METADATA_DB_NAME))
+ {
+ final MessageMetaDataBinding binding = new MessageMetaDataBinding();
+ CursorOperation databaseOperation = new CursorOperation()
+ {
+
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ StorableMessageMetaData metaData = binding.entryToObject(value);
+
+ // get message id
+ Long messageId = LongBinding.entryToLong(key);
+
+ // ONLY copy data if message is delivered to existing queue
+ if (messagesToDiscard.contains(messageId))
+ {
+ return;
+ }
+ DatabaseEntry newValue = new DatabaseEntry();
+ binding.objectToEntry(metaData, newValue);
+
+ targetDatabase.put(transaction, key, newValue);
+ targetDatabase.put(transaction, key, newValue);
+ deleteCurrent();
+
+ }
+ };
+
+ new DatabaseTemplate(environment, OLD_METADATA_DB_NAME, NEW_METADATA_DB_NAME, transaction)
+ .run(databaseOperation);
+ environment.removeDatabase(transaction, OLD_METADATA_DB_NAME);
+ _logger.info(databaseOperation.getRowCount() + " Message MetaData entries");
+ }
+ }
+
+ private void upgradeContent(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler, final Set<Long> messagesToDiscard, Transaction transaction)
+ {
+ _logger.info("Message Contents");
+ if (environment.getDatabaseNames().contains(OLD_CONTENT_DB_NAME))
+ {
+ final MessageContentKeyBinding keyBinding = new MessageContentKeyBinding();
+ final ContentBinding contentBinding = new ContentBinding();
+ CursorOperation cursorOperation = new CursorOperation()
+ {
+ private long _prevMsgId = -1;
+ private int _bytesSeenSoFar;
+
+ @Override
+ public void processEntry(Database sourceDatabase, Database targetDatabase, Transaction transaction,
+ DatabaseEntry key, DatabaseEntry value)
+ {
+ // determine the msgId of the current entry
+ MessageContentKey contentKey = keyBinding.entryToObject(key);
+ long msgId = contentKey.getMessageId();
+
+ // ONLY copy data if message is delivered to existing queue
+ if (messagesToDiscard.contains(msgId))
+ {
+ return;
+ }
+ // if this is a new message, restart the byte offset count.
+ if (_prevMsgId != msgId)
+ {
+ _bytesSeenSoFar = 0;
+ }
+
+ // determine the content size
+ ByteBuffer content = contentBinding.entryToObject(value);
+ int contentSize = content.limit();
+
+ // create the new key: id + previously seen data count
+ MessageContentKey newKey = new MessageContentKey(msgId, _bytesSeenSoFar);
+ DatabaseEntry newKeyEntry = new DatabaseEntry();
+ keyBinding.objectToEntry(newKey, newKeyEntry);
+
+ DatabaseEntry newValueEntry = new DatabaseEntry();
+ contentBinding.objectToEntry(content, newValueEntry);
+
+ targetDatabase.put(null, newKeyEntry, newValueEntry);
+
+ _prevMsgId = msgId;
+ _bytesSeenSoFar += contentSize;
+ }
+ };
+ new DatabaseTemplate(environment, OLD_CONTENT_DB_NAME, NEW_CONTENT_DB_NAME, transaction).run(cursorOperation);
+ environment.removeDatabase(transaction, OLD_CONTENT_DB_NAME);
+ _logger.info(cursorOperation.getRowCount() + " Message Content entries");
+ }
+ }
+
+ /**
+ * For all databases which haven't been otherwise upgraded, we still need to
+ * rename them from _v4 to _v5
+ */
+ private void renameRemainingDatabases(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler, Transaction transaction)
+ {
+ for (String dbName : environment.getDatabaseNames())
+ {
+ if (dbName.endsWith("_v4"))
+ {
+ String newName = dbName.substring(0, dbName.length() - 3) + "_v5";
+ _logger.info("Renaming " + dbName + " into " + newName);
+ environment.renameDatabase(transaction, dbName, newName);
+ }
+ }
+
+ }
+
+ private void addBindingToDatabase(final BindingTuple bindingTuple, Database targetDatabase, Transaction transaction,
+ AMQShortString queueName, AMQShortString exchangeName, AMQShortString routingKey, FieldTable arguments)
+ {
+
+ DatabaseEntry newKey = new DatabaseEntry();
+
+ bindingTuple.objectToEntry(new BindingRecord(exchangeName, queueName, routingKey, arguments), newKey);
+
+ DatabaseEntry newValue = new DatabaseEntry();
+ ByteBinding.byteToEntry((byte) 0, newValue);
+
+ targetDatabase.put(transaction, newKey, newValue);
+ }
+
+ private static final class ExchangeRecord
+ {
+ private final AMQShortString _name;
+ private final AMQShortString _type;
+
+ private ExchangeRecord(final AMQShortString name, final AMQShortString type)
+ {
+ _name = name;
+ _type = type;
+ }
+
+ public AMQShortString getName()
+ {
+ return _name;
+ }
+
+ public AMQShortString getType()
+ {
+ return _type;
+ }
+ }
+
+ private static final class ExchangeRecordBinding extends TupleBinding<ExchangeRecord>
+ {
+
+ @Override
+ public ExchangeRecord entryToObject(final TupleInput input)
+ {
+ return new ExchangeRecord(AMQShortStringEncoding.readShortString(input),
+ AMQShortStringEncoding.readShortString(input));
+ }
+
+ @Override
+ public void objectToEntry(final ExchangeRecord object, final TupleOutput output)
+ {
+ AMQShortStringEncoding.writeShortString(object.getName(), output);
+ AMQShortStringEncoding.writeShortString(object.getType(), output);
+ output.writeBoolean(false);
+ }
+ }
+
+ private static final class PartialBindingRecord
+ {
+ private final AMQShortString _exchangeName;
+ private final AMQShortString _queueName;
+
+ private PartialBindingRecord(final AMQShortString name, final AMQShortString type)
+ {
+ _exchangeName = name;
+ _queueName = type;
+ }
+
+ public AMQShortString getExchangeName()
+ {
+ return _exchangeName;
+ }
+
+ public AMQShortString getQueueName()
+ {
+ return _queueName;
+ }
+ }
+
+ private static final class PartialBindingRecordBinding extends TupleBinding<PartialBindingRecord>
+ {
+
+ @Override
+ public PartialBindingRecord entryToObject(final TupleInput input)
+ {
+ return new PartialBindingRecord(AMQShortStringEncoding.readShortString(input),
+ AMQShortStringEncoding.readShortString(input));
+ }
+
+ @Override
+ public void objectToEntry(final PartialBindingRecord object, final TupleOutput output)
+ {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ static final class QueueRecord
+ {
+ private final AMQShortString _queueName;
+ private final AMQShortString _owner;
+ private final FieldTable _arguments;
+ private final boolean _exclusive;
+
+ public QueueRecord(AMQShortString queueName, AMQShortString owner, boolean exclusive, FieldTable arguments)
+ {
+ _queueName = queueName;
+ _owner = owner;
+ _exclusive = exclusive;
+ _arguments = arguments;
+ }
+
+ public AMQShortString getNameShortString()
+ {
+ return _queueName;
+ }
+
+ public AMQShortString getOwner()
+ {
+ return _owner;
+ }
+
+ public boolean isExclusive()
+ {
+ return _exclusive;
+ }
+
+ public FieldTable getArguments()
+ {
+ return _arguments;
+ }
+ }
+
+ static final class QueueRecordBinding extends TupleBinding<QueueRecord>
+ {
+ private final List<AMQShortString> _durableSubNames;
+
+ QueueRecordBinding(final List<AMQShortString> durableSubNames)
+ {
+ _durableSubNames = durableSubNames;
+ }
+
+ @Override
+ public QueueRecord entryToObject(final TupleInput input)
+ {
+ AMQShortString name = AMQShortStringEncoding.readShortString(input);
+ AMQShortString owner = AMQShortStringEncoding.readShortString(input);
+ FieldTable arguments = FieldTableEncoding.readFieldTable(input);
+ boolean exclusive = input.available() > 0 && input.readBoolean();
+ exclusive = exclusive || _durableSubNames.contains(name);
+
+ return new QueueRecord(name, owner, exclusive, arguments);
+
+ }
+
+ @Override
+ public void objectToEntry(final QueueRecord record, final TupleOutput output)
+ {
+ AMQShortStringEncoding.writeShortString(record.getNameShortString(), output);
+ AMQShortStringEncoding.writeShortString(record.getOwner(), output);
+ FieldTableEncoding.writeFieldTable(record.getArguments(), output);
+ output.writeBoolean(record.isExclusive());
+
+ }
+ }
+
+ static final class MessageMetaDataBinding extends TupleBinding<StorableMessageMetaData>
+ {
+
+ @Override
+ public MessageMetaData entryToObject(final TupleInput input)
+ {
+ try
+ {
+ final MessagePublishInfo publishBody = readMessagePublishInfo(input);
+ final ContentHeaderBody contentHeaderBody = readContentHeaderBody(input);
+ final int contentChunkCount = input.readInt();
+
+ return new MessageMetaData(publishBody, contentHeaderBody, contentChunkCount);
+ }
+ catch (Exception e)
+ {
+ _logger.error("Error converting entry to object: " + e, e);
+ // annoyingly just have to return null since we cannot throw
+ return null;
+ }
+ }
+
+ private MessagePublishInfo readMessagePublishInfo(TupleInput tupleInput)
+ {
+
+ final AMQShortString exchange = AMQShortStringEncoding.readShortString(tupleInput);
+ final AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
+ final boolean mandatory = tupleInput.readBoolean();
+ final boolean immediate = tupleInput.readBoolean();
+
+ return new MessagePublishInfo()
+ {
+
+ public AMQShortString getExchange()
+ {
+ return exchange;
+ }
+
+ public void setExchange(AMQShortString exchange)
+ {
+
+ }
+
+ public boolean isImmediate()
+ {
+ return immediate;
+ }
+
+ public boolean isMandatory()
+ {
+ return mandatory;
+ }
+
+ public AMQShortString getRoutingKey()
+ {
+ return routingKey;
+ }
+ };
+
+ }
+
+ private ContentHeaderBody readContentHeaderBody(TupleInput tupleInput) throws AMQFrameDecodingException,
+ AMQProtocolVersionException
+ {
+ int bodySize = tupleInput.readInt();
+ byte[] underlying = new byte[bodySize];
+ tupleInput.readFast(underlying);
+
+ try
+ {
+ return ContentHeaderBody.createFromBuffer(new DataInputStream(new ByteArrayInputStream(underlying)),
+ bodySize);
+ }
+ catch (IOException e)
+ {
+ throw new AMQFrameDecodingException(null, e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void objectToEntry(final StorableMessageMetaData metaData, final TupleOutput output)
+ {
+ final int bodySize = 1 + metaData.getStorableSize();
+ byte[] underlying = new byte[bodySize];
+ underlying[0] = (byte) metaData.getType().ordinal();
+ java.nio.ByteBuffer buf = java.nio.ByteBuffer.wrap(underlying);
+ buf.position(1);
+ buf = buf.slice();
+
+ metaData.writeToBuffer(0, buf);
+ output.writeInt(bodySize);
+ output.writeFast(underlying);
+ }
+ }
+
+ static final class MessageContentKey
+ {
+ private long _messageId;
+ private int _chunk;
+
+ public MessageContentKey(long messageId, int chunkNo)
+ {
+ _messageId = messageId;
+ _chunk = chunkNo;
+ }
+
+ public int getChunk()
+ {
+ return _chunk;
+ }
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+
+ }
+
+ static final class MessageContentKeyBinding extends TupleBinding<MessageContentKey>
+ {
+
+ public MessageContentKey entryToObject(TupleInput tupleInput)
+ {
+ long messageId = tupleInput.readLong();
+ int chunk = tupleInput.readInt();
+ return new MessageContentKey(messageId, chunk);
+ }
+
+ public void objectToEntry(MessageContentKey object, TupleOutput tupleOutput)
+ {
+ final MessageContentKey mk = object;
+ tupleOutput.writeLong(mk.getMessageId());
+ tupleOutput.writeInt(mk.getChunk());
+ }
+
+ }
+
+ static final class ContentBinding extends TupleBinding<ByteBuffer>
+ {
+ public ByteBuffer entryToObject(TupleInput tupleInput)
+ {
+ final int size = tupleInput.readInt();
+ byte[] underlying = new byte[size];
+ tupleInput.readFast(underlying);
+ return ByteBuffer.wrap(underlying);
+ }
+
+ public void objectToEntry(ByteBuffer src, TupleOutput tupleOutput)
+ {
+ src = src.slice();
+
+ byte[] chunkData = new byte[src.limit()];
+ src.duplicate().get(chunkData);
+
+ tupleOutput.writeInt(chunkData.length);
+ tupleOutput.writeFast(chunkData);
+ }
+ }
+
+ static final class QueueEntryKey
+ {
+ private AMQShortString _queueName;
+ private long _messageId;
+
+ public QueueEntryKey(AMQShortString queueName, long messageId)
+ {
+ _queueName = queueName;
+ _messageId = messageId;
+ }
+
+ public AMQShortString getQueueName()
+ {
+ return _queueName;
+ }
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+
+ }
+
+ static final class QueueEntryKeyBinding extends TupleBinding<QueueEntryKey>
+ {
+ public QueueEntryKey entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
+ long messageId = tupleInput.readLong();
+ return new QueueEntryKey(queueName, messageId);
+ }
+
+ public void objectToEntry(QueueEntryKey mk, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(mk.getQueueName(), tupleOutput);
+ tupleOutput.writeLong(mk.getMessageId());
+ }
+ }
+
+ static final class BindingRecord extends Object
+ {
+ private final AMQShortString _exchangeName;
+ private final AMQShortString _queueName;
+ private final AMQShortString _routingKey;
+ private final FieldTable _arguments;
+
+ public BindingRecord(AMQShortString exchangeName, AMQShortString queueName, AMQShortString routingKey,
+ FieldTable arguments)
+ {
+ _exchangeName = exchangeName;
+ _queueName = queueName;
+ _routingKey = routingKey;
+ _arguments = arguments;
+ }
+
+ public AMQShortString getExchangeName()
+ {
+ return _exchangeName;
+ }
+
+ public AMQShortString getQueueName()
+ {
+ return _queueName;
+ }
+
+ public AMQShortString getRoutingKey()
+ {
+ return _routingKey;
+ }
+
+ public FieldTable getArguments()
+ {
+ return _arguments;
+ }
+
+ }
+
+ static final class BindingTuple extends TupleBinding<BindingRecord>
+ {
+ public BindingRecord entryToObject(TupleInput tupleInput)
+ {
+ AMQShortString exchangeName = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString queueName = AMQShortStringEncoding.readShortString(tupleInput);
+ AMQShortString routingKey = AMQShortStringEncoding.readShortString(tupleInput);
+
+ FieldTable arguments = FieldTableEncoding.readFieldTable(tupleInput);
+
+ return new BindingRecord(exchangeName, queueName, routingKey, arguments);
+ }
+
+ public void objectToEntry(BindingRecord binding, TupleOutput tupleOutput)
+ {
+ AMQShortStringEncoding.writeShortString(binding.getExchangeName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(binding.getQueueName(), tupleOutput);
+ AMQShortStringEncoding.writeShortString(binding.getRoutingKey(), tupleOutput);
+
+ FieldTableEncoding.writeFieldTable(binding.getArguments(), tupleOutput);
+ }
+
+ }
+}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeFrom5To6.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,372 @@
+/*
+ *
+ * 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.berkeleydb.upgrade;
+
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeInteractionResponse.ABORT;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeInteractionResponse.NO;
+import static org.apache.qpid.server.store.berkeleydb.upgrade.UpgradeInteractionResponse.YES;
+
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.logging.LogSubject;
+
+import com.sleepycat.bind.tuple.LongBinding;
+import com.sleepycat.bind.tuple.TupleBinding;
+import com.sleepycat.bind.tuple.TupleInput;
+import com.sleepycat.bind.tuple.TupleOutput;
+import com.sleepycat.je.Cursor;
+import com.sleepycat.je.CursorConfig;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.LockMode;
+import com.sleepycat.je.OperationStatus;
+import com.sleepycat.je.Transaction;
+
+public class UpgradeFrom5To6 extends AbstractStoreUpgrade
+{
+
+ private static final Logger _logger = Logger.getLogger(UpgradeFrom5To6.class);
+
+ private static final String OLD_CONTENT_DB_NAME = "messageContentDb_v5";
+ private static final String NEW_CONTENT_DB_NAME = "MESSAGE_CONTENT";
+ private static final String META_DATA_DB_NAME = "messageMetaDataDb_v5";
+
+ private static final String NEW_DB_NAMES[] = { "EXCHANGES", "QUEUES", "QUEUE_BINDINGS", "DELIVERIES",
+ "MESSAGE_METADATA", NEW_CONTENT_DB_NAME, "BRIDGES", "LINKS", "XIDS" };
+ private static final String OLD_DB_NAMES[] = { "exchangeDb_v5", "queueDb_v5", "queueBindingsDb_v5", "deliveryDb_v5",
+ META_DATA_DB_NAME, OLD_CONTENT_DB_NAME, "bridges_v5", "links_v5", "xids_v5" };
+
+ public void performUpgrade(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler) throws DatabaseException, AMQStoreException
+ {
+ _logger.info("Starting store upgrade from version 5");
+ Transaction transaction = null;
+ try
+ {
+ reportStarting(environment, OLD_DB_NAMES, USER_FRIENDLY_NAMES);
+ transaction = environment.beginTransaction(null, null);
+ performUpgradeInternal(logSubject, environment, handler, transaction);
+ transaction.commit();
+ reportFinished(environment, NEW_DB_NAMES, USER_FRIENDLY_NAMES);
+ }
+ catch (Exception e)
+ {
+ transaction.abort();
+ if (e instanceof DatabaseException)
+ {
+ throw (DatabaseException) e;
+ }
+ else if (e instanceof AMQStoreException)
+ {
+ throw (AMQStoreException) e;
+ }
+ else
+ {
+ throw new AMQStoreException("Unexpected exception", e);
+ }
+ }
+ }
+
+ /**
+ * Upgrades from a v5 database to a v6 database
+ *
+ * v6 is the first "new style" schema where we don't version every table, and the upgrade is re-runnable
+ *
+ * Change in this version:
+ *
+ * Message content is moved from the database messageContentDb_v5 to MESSAGE_CONTENT.
+ * The structure of the database changes from
+ * ( message-id: long, chunk-id: int ) -> ( size: int, byte[] data )
+ * to
+ * ( message-id: long) -> ( byte[] data )
+ *
+ * That is we keep only one record per message, which contains all the message content
+ */
+ public void performUpgradeInternal(final LogSubject logSubject, final Environment environment,
+ final UpgradeInteractionHandler handler, final Transaction transaction) throws AMQStoreException
+ {
+ _logger.info("Message Contents");
+ if (environment.getDatabaseNames().contains(OLD_CONTENT_DB_NAME))
+ {
+ DatabaseRunnable contentOperation = new DatabaseRunnable()
+ {
+ @Override
+ public void run(final Database oldContentDatabase, final Database newContentDatabase,
+ Transaction contentTransaction)
+ {
+ CursorOperation metaDataDatabaseOperation = new CursorOperation()
+ {
+
+ @Override
+ public void processEntry(Database metadataDatabase, Database notUsed,
+ Transaction metaDataTransaction, DatabaseEntry key, DatabaseEntry value)
+ {
+ long messageId = LongBinding.entryToLong(key);
+ upgradeMessage(messageId, oldContentDatabase, newContentDatabase, handler, metaDataTransaction,
+ metadataDatabase);
+ }
+ };
+ new DatabaseTemplate(environment, META_DATA_DB_NAME, contentTransaction).run(metaDataDatabaseOperation);
+ _logger.info(metaDataDatabaseOperation.getRowCount() + " Message Content Entries");
+ }
+ };
+ new DatabaseTemplate(environment, OLD_CONTENT_DB_NAME, NEW_CONTENT_DB_NAME, transaction).run(contentOperation);
+ }
+ renameDatabases(environment, transaction);
+ }
+
+ private void renameDatabases(Environment environment, Transaction transaction)
+ {
+ List<String> databases = environment.getDatabaseNames();
+ for (int i = 0; i < OLD_DB_NAMES.length; i++)
+ {
+ String oldName = OLD_DB_NAMES[i];
+ String newName = NEW_DB_NAMES[i];
+ if (databases.contains(oldName))
+ {
+ _logger.info("Renaming " + oldName + " into " + newName);
+ environment.renameDatabase(transaction, oldName, newName);
+ }
+ }
+ }
+
+ /**
+ * Upgrade an individual message, that is read all the data from the old
+ * database, consolidate it into a single byte[] and then (in a transaction)
+ * remove the record from the old database and add the corresponding record
+ * to the new database
+ */
+ private void upgradeMessage(final long messageId, final Database oldDatabase, final Database newDatabase,
+ final UpgradeInteractionHandler handler, Transaction txn, Database oldMetadataDatabase)
+ {
+ SortedMap<Integer, byte[]> messageData = getMessageData(messageId, oldDatabase);
+ byte[] consolidatedData = new byte[0];
+ for (Map.Entry<Integer, byte[]> entry : messageData.entrySet())
+ {
+ int offset = entry.getKey();
+ if (offset != consolidatedData.length)
+ {
+ String message;
+ if (offset < consolidatedData.length)
+ {
+ message = "Missing data in message id " + messageId + " between offset " + consolidatedData.length
+ + " and " + offset + ". ";
+ }
+ else
+ {
+ message = "Duplicate data in message id " + messageId + " between offset " + offset + " and "
+ + consolidatedData.length + ". ";
+ }
+ UpgradeInteractionResponse action = handler.requireResponse(message
+ + "Do you wish do recover as much of this message as "
+ + "possible (answering NO will delete the message)?", ABORT, YES, NO, ABORT);
+
+ switch (action)
+ {
+ case YES:
+ byte[] oldData = consolidatedData;
+ consolidatedData = new byte[offset];
+ System.arraycopy(oldData, 0, consolidatedData, 0, Math.min(oldData.length, consolidatedData.length));
+ break;
+ case NO:
+ DatabaseEntry key = new DatabaseEntry();
+ LongBinding.longToEntry(messageId, key);
+ oldMetadataDatabase.delete(txn, key);
+ return;
+ case ABORT:
+ _logger.error(message);
+ throw new RuntimeException("Unable to upgrade message " + messageId);
+ }
+
+ }
+ byte[] data = new byte[consolidatedData.length + entry.getValue().length];
+ System.arraycopy(consolidatedData, 0, data, 0, consolidatedData.length);
+ System.arraycopy(entry.getValue(), 0, data, offset, entry.getValue().length);
+ consolidatedData = data;
+ }
+
+ CompoundKeyBinding binding = new CompoundKeyBinding();
+ for (int offset : messageData.keySet())
+ {
+ DatabaseEntry key = new DatabaseEntry();
+ binding.objectToEntry(new CompoundKey(messageId, offset), key);
+ oldDatabase.delete(txn, key);
+ }
+ DatabaseEntry key = new DatabaseEntry();
+ LongBinding.longToEntry(messageId, key);
+ NewDataBinding dataBinding = new NewDataBinding();
+ DatabaseEntry value = new DatabaseEntry();
+ dataBinding.objectToEntry(consolidatedData, value);
+
+ newDatabase.put(txn, key, value);
+ }
+
+ /**
+ * @return a (sorted) map of offset -> data for the given message id
+ */
+ private SortedMap<Integer, byte[]> getMessageData(final long messageId, final Database oldDatabase)
+ {
+ TreeMap<Integer, byte[]> data = new TreeMap<Integer, byte[]>();
+
+ Cursor cursor = oldDatabase.openCursor(null, CursorConfig.READ_COMMITTED);
+ try
+ {
+ DatabaseEntry contentKeyEntry = new DatabaseEntry();
+ DatabaseEntry value = new DatabaseEntry();
+ CompoundKeyBinding binding = new CompoundKeyBinding();
+ binding.objectToEntry(new CompoundKey(messageId, 0), contentKeyEntry);
+
+ OperationStatus status = cursor.getSearchKeyRange(contentKeyEntry, value, LockMode.DEFAULT);
+ OldDataBinding dataBinding = new OldDataBinding();
+
+ while (status == OperationStatus.SUCCESS)
+ {
+ CompoundKey compoundKey = binding.entryToObject(contentKeyEntry);
+ long id = compoundKey.getMessageId();
+
+ if (id != messageId)
+ {
+ // we have exhausted all chunks for this message id, break
+ break;
+ }
+
+ int offsetInMessage = compoundKey.getOffset();
+ OldDataValue dataValue = dataBinding.entryToObject(value);
+ data.put(offsetInMessage, dataValue.getData());
+
+ status = cursor.getNext(contentKeyEntry, value, LockMode.DEFAULT);
+ }
+ }
+ finally
+ {
+ cursor.close();
+ }
+
+ return data;
+ }
+
+ static final class CompoundKey
+ {
+ public final long _messageId;
+ public final int _offset;
+
+ public CompoundKey(final long messageId, final int offset)
+ {
+ _messageId = messageId;
+ _offset = offset;
+ }
+
+ public long getMessageId()
+ {
+ return _messageId;
+ }
+
+ public int getOffset()
+ {
+ return _offset;
+ }
+ }
+
+ static final class CompoundKeyBinding extends TupleBinding<CompoundKey>
+ {
+
+ @Override
+ public CompoundKey entryToObject(final TupleInput input)
+ {
+ return new CompoundKey(input.readLong(), input.readInt());
+ }
+
+ @Override
+ public void objectToEntry(final CompoundKey object, final TupleOutput output)
+ {
+ output.writeLong(object._messageId);
+ output.writeInt(object._offset);
+ }
+ }
+
+ static final class OldDataValue
+ {
+ private final int _size;
+ private final byte[] _data;
+
+ private OldDataValue(final int size, final byte[] data)
+ {
+ _size = size;
+ _data = data;
+ }
+
+ public int getSize()
+ {
+ return _size;
+ }
+
+ public byte[] getData()
+ {
+ return _data;
+ }
+ }
+
+ static final class OldDataBinding extends TupleBinding<OldDataValue>
+ {
+
+ @Override
+ public OldDataValue entryToObject(final TupleInput input)
+ {
+ int size = input.readInt();
+ byte[] data = new byte[size];
+ input.read(data);
+ return new OldDataValue(size, data);
+ }
+
+ @Override
+ public void objectToEntry(OldDataValue value, final TupleOutput output)
+ {
+ output.writeInt(value.getSize());
+ output.write(value.getData());
+ }
+ }
+
+ static final class NewDataBinding extends TupleBinding<byte[]>
+ {
+
+ @Override
+ public byte[] entryToObject(final TupleInput input)
+ {
+ byte[] data = new byte[input.available()];
+ input.read(data);
+ return data;
+ }
+
+ @Override
+ public void objectToEntry(final byte[] data, final TupleOutput output)
+ {
+ output.write(data);
+ }
+ }
+
+}
\ No newline at end of file
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionHandler.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionHandler.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionHandler.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTupleBindingFactory.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionHandler.java Tue Mar 27 11:04:02 2012
@@ -18,28 +18,20 @@
* under the License.
*
*/
-package org.apache.qpid.server.store.berkeleydb.tuples;
+package org.apache.qpid.server.store.berkeleydb.upgrade;
-import com.sleepycat.bind.tuple.TupleBinding;
-
-import org.apache.qpid.server.store.berkeleydb.records.BindingRecord;
-
-public class BindingTupleBindingFactory extends TupleBindingFactory<BindingRecord>
+public interface UpgradeInteractionHandler
{
- public BindingTupleBindingFactory(int version)
- {
- super(version);
- }
+ UpgradeInteractionResponse requireResponse(String question, UpgradeInteractionResponse defaultResponse,
+ UpgradeInteractionResponse... possibleResponses);
- public TupleBinding<BindingRecord> getInstance()
+ public static final UpgradeInteractionHandler DEFAULT_HANDLER = new UpgradeInteractionHandler()
{
- switch (getVersion())
+ public UpgradeInteractionResponse requireResponse(final String question,
+ final UpgradeInteractionResponse defaultResponse,
+ final UpgradeInteractionResponse... possibleResponses)
{
- default:
- case 5:
- //no change from v4
- case 4:
- return new BindingTuple_4();
+ return defaultResponse;
}
- }
+ };
}
Copied: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionResponse.java (from r1305371, qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionResponse.java?p2=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionResponse.java&p1=qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java&r1=1305371&r2=1305809&rev=1305809&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuples/BindingTuple.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgradeInteractionResponse.java Tue Mar 27 11:04:02 2012
@@ -18,8 +18,11 @@
* under the License.
*
*/
-package org.apache.qpid.server.store.berkeleydb.tuples;
+package org.apache.qpid.server.store.berkeleydb.upgrade;
-public interface BindingTuple
+public enum UpgradeInteractionResponse
{
+ YES,
+ NO,
+ ABORT
}
Added: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java?rev=1305809&view=auto
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java (added)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java Tue Mar 27 11:04:02 2012
@@ -0,0 +1,178 @@
+/*
+ *
+ * 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.berkeleydb.upgrade;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+
+import com.sleepycat.bind.tuple.IntegerBinding;
+import com.sleepycat.bind.tuple.LongBinding;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseConfig;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.LockMode;
+import com.sleepycat.je.OperationStatus;
+
+public class Upgrader
+{
+ static final String VERSION_DB_NAME = "VERSION";
+
+ private Environment _environment;
+ private LogSubject _logSubject;
+
+ public Upgrader(Environment environment, LogSubject logSubject)
+ {
+ _environment = environment;
+ _logSubject = logSubject;
+ }
+
+ public void upgradeIfNecessary() throws AMQStoreException
+ {
+ boolean isEmpty = _environment.getDatabaseNames().isEmpty();
+ DatabaseConfig dbConfig = new DatabaseConfig();
+ dbConfig.setTransactional(true);
+ dbConfig.setAllowCreate(true);
+
+ Database versionDb = null;
+ try
+ {
+ versionDb = _environment.openDatabase(null, VERSION_DB_NAME, dbConfig);
+
+ if(versionDb.count() == 0L)
+ {
+ int sourceVersion = isEmpty ? BDBMessageStore.VERSION: identifyOldStoreVersion();
+ DatabaseEntry key = new DatabaseEntry();
+ IntegerBinding.intToEntry(sourceVersion, key);
+ DatabaseEntry value = new DatabaseEntry();
+ LongBinding.longToEntry(System.currentTimeMillis(), value);
+
+ versionDb.put(null, key, value);
+ }
+
+ int version = getSourceVersion(versionDb);
+
+ performUpgradeFromVersion(version, versionDb);
+ }
+ finally
+ {
+ if (versionDb != null)
+ {
+ versionDb.close();
+ }
+ }
+ }
+
+ int getSourceVersion(Database versionDb)
+ {
+ int version = BDBMessageStore.VERSION + 1;
+ OperationStatus result;
+
+ do
+ {
+ version--;
+ DatabaseEntry key = new DatabaseEntry();
+ IntegerBinding.intToEntry(version, key);
+ DatabaseEntry value = new DatabaseEntry();
+
+ result = versionDb.get(null, key, value, LockMode.READ_COMMITTED);
+ }
+ while(result == OperationStatus.NOTFOUND);
+ return version;
+ }
+
+ void performUpgradeFromVersion(int sourceVersion, Database versionDb)
+ throws AMQStoreException
+ {
+ while(sourceVersion != BDBMessageStore.VERSION)
+ {
+ upgrade(sourceVersion, ++sourceVersion);
+ DatabaseEntry key = new DatabaseEntry();
+ IntegerBinding.intToEntry(sourceVersion, key);
+ DatabaseEntry value = new DatabaseEntry();
+ LongBinding.longToEntry(System.currentTimeMillis(), value);
+ versionDb.put(null, key, value);
+ }
+ }
+
+ void upgrade(final int fromVersion, final int toVersion) throws AMQStoreException
+ {
+ try
+ {
+ @SuppressWarnings("unchecked")
+ Class<StoreUpgrade> upgradeClass =
+ (Class<StoreUpgrade>) Class.forName("org.apache.qpid.server.store.berkeleydb.upgrade."
+ + "UpgradeFrom"+fromVersion+"To"+toVersion);
+ Constructor<StoreUpgrade> ctr = upgradeClass.getConstructor();
+ StoreUpgrade upgrade = ctr.newInstance();
+ upgrade.performUpgrade(_logSubject, _environment, UpgradeInteractionHandler.DEFAULT_HANDLER);
+ }
+ catch (ClassNotFoundException e)
+ {
+ throw new AMQStoreException("Unable to upgrade BDB data store from version " + fromVersion + " to version"
+ + toVersion, e);
+ }
+ catch (NoSuchMethodException e)
+ {
+ throw new AMQStoreException("Unable to upgrade BDB data store from version " + fromVersion + " to version"
+ + toVersion, e);
+ }
+ catch (InvocationTargetException e)
+ {
+ throw new AMQStoreException("Unable to upgrade BDB data store from version " + fromVersion + " to version"
+ + toVersion, e);
+ }
+ catch (InstantiationException e)
+ {
+ throw new AMQStoreException("Unable to upgrade BDB data store from version " + fromVersion + " to version"
+ + toVersion, e);
+ }
+ catch (IllegalAccessException e)
+ {
+ throw new AMQStoreException("Unable to upgrade BDB data store from version " + fromVersion + " to version"
+ + toVersion, e);
+ }
+ }
+
+ private int identifyOldStoreVersion() throws DatabaseException
+ {
+ int version = 0;
+ for (String databaseName : _environment.getDatabaseNames())
+ {
+ if (databaseName.contains("_v"))
+ {
+ int versionIndex = databaseName.indexOf("_v");
+ if (versionIndex == -1)
+ {
+ versionIndex = 1;
+ }
+ version = Integer.parseInt(databaseName.substring(versionIndex + 2));
+ break;
+ }
+ }
+ return version;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org