QPID-4983 : [Java Broker] Move store implementations to broker plugins

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1501682 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 999 insertions, 0 deletions

diff --git a/qpid/java/broker-plugins/jdbc-store/build.xml b/qpid/java/broker-plugins/jdbc-store/build.xml
new file mode 100644
index 0000000000..de6ec59845
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/build.xml
@@ -0,0 +1,31 @@
+<!--
+ - 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.
+ -->
+<project name="Qpid Broker-Plugins JDBC Store" default="build">
+    <property name="module.depends" value="common broker" />
+    <property name="module.test.depends" value="common/tests broker/tests" />
+
+    <property name="module.genpom" value="true"/>
+    <property name="module.genpom.args" value="-Sqpid-common=provided -Sqpid-broker=provided"/>
+
+    <property name="broker.plugin" value="true"/>
+
+    <import file="../../module.xml" />
+
+    <target name="bundle" depends="bundle-tasks"/>
+</project>
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/DefaultConnectionProvider.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/DefaultConnectionProvider.java
new file mode 100644
index 0000000000..7945ae3b46
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/DefaultConnectionProvider.java
@@ -0,0 +1,46 @@
+/*
+ *
+ * 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.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+
+class DefaultConnectionProvider implements ConnectionProvider
+{
+    private final String _connectionUrl;
+
+    public DefaultConnectionProvider(String connectionUrl)
+    {
+        _connectionUrl = connectionUrl;
+    }
+
+    @Override
+    public Connection getConnection() throws SQLException
+    {
+        return DriverManager.getConnection(_connectionUrl);
+    }
+
+    @Override
+    public void close() throws SQLException
+    {
+    }
+}
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/DefaultConnectionProviderFactory.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/DefaultConnectionProviderFactory.java
new file mode 100644
index 0000000000..8fc7de12d0
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/DefaultConnectionProviderFactory.java
@@ -0,0 +1,42 @@
+/*
+ *
+ * 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.jdbc;
+
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.plugin.JDBCConnectionProviderFactory;
+
+public class DefaultConnectionProviderFactory implements JDBCConnectionProviderFactory
+{
+
+    @Override
+    public String getType()
+    {
+        return "NONE";
+    }
+
+    @Override
+    public ConnectionProvider getConnectionProvider(String connectionUrl,
+                                                    VirtualHost virtualHost)
+    {
+        return new DefaultConnectionProvider(connectionUrl);
+    }
+
+}
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCMessageStore.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCMessageStore.java
new file mode 100644
index 0000000000..f8d93536bb
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCMessageStore.java
@@ -0,0 +1,462 @@
+/*
+*
+* 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.jdbc;
+
+
+import java.sql.Blob;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.plugin.JDBCConnectionProviderFactory;
+import org.apache.qpid.server.store.AbstractJDBCMessageStore;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.Transaction;
+
+/**
+ * An implementation of a {@link org.apache.qpid.server.store.MessageStore} that uses a JDBC database as the persistence
+ * mechanism.
+ *
+ */
+public class JDBCMessageStore extends AbstractJDBCMessageStore implements MessageStore
+{
+
+    private static final Logger _logger = Logger.getLogger(JDBCMessageStore.class);
+
+
+    public static final String TYPE = "JDBC";
+    public static final String CONNECTION_URL = "connectionURL";
+
+    protected String _connectionURL;
+    private ConnectionProvider _connectionProvider;
+
+
+    private static class JDBCDetails
+    {
+        private final String _vendor;
+        private String _blobType;
+        private String _varBinaryType;
+        private String _bigintType;
+        private boolean _useBytesMethodsForBlob;
+
+        private JDBCDetails(String vendor,
+                            String blobType,
+                            String varBinaryType,
+                            String bigIntType,
+                            boolean useBytesMethodsForBlob)
+        {
+            _vendor = vendor;
+            setBlobType(blobType);
+            setVarBinaryType(varBinaryType);
+            setBigintType(bigIntType);
+            setUseBytesMethodsForBlob(useBytesMethodsForBlob);
+        }
+
+        @Override
+        public boolean equals(Object o)
+        {
+            if (this == o)
+            {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass())
+            {
+                return false;
+            }
+
+            JDBCDetails that = (JDBCDetails) o;
+
+            if (!getVendor().equals(that.getVendor()))
+            {
+                return false;
+            }
+
+            return true;
+        }
+
+        @Override
+        public int hashCode()
+        {
+            return getVendor().hashCode();
+        }
+
+        @Override
+        public String toString()
+        {
+            return "JDBCDetails{" +
+                   "vendor='" + getVendor() + '\'' +
+                   ", blobType='" + getBlobType() + '\'' +
+                   ", varBinaryType='" + getVarBinaryType() + '\'' +
+                   ", bigIntType='" + getBigintType() + '\'' +
+                   ", useBytesMethodsForBlob=" + isUseBytesMethodsForBlob() +
+                   '}';
+        }
+
+        public String getVendor()
+        {
+            return _vendor;
+        }
+
+        public String getBlobType()
+        {
+            return _blobType;
+        }
+
+        public void setBlobType(String blobType)
+        {
+            _blobType = blobType;
+        }
+
+        public String getVarBinaryType()
+        {
+            return _varBinaryType;
+        }
+
+        public void setVarBinaryType(String varBinaryType)
+        {
+            _varBinaryType = varBinaryType;
+        }
+
+        public boolean isUseBytesMethodsForBlob()
+        {
+            return _useBytesMethodsForBlob;
+        }
+
+        public void setUseBytesMethodsForBlob(boolean useBytesMethodsForBlob)
+        {
+            _useBytesMethodsForBlob = useBytesMethodsForBlob;
+        }
+
+        public String getBigintType()
+        {
+            return _bigintType;
+        }
+
+        public void setBigintType(String bigintType)
+        {
+            _bigintType = bigintType;
+        }
+    }
+
+    private static JDBCDetails DERBY_DETAILS =
+            new JDBCDetails("derby",
+                            "blob",
+                            "varchar(%d) for bit data",
+                            "bigint",
+                            false);
+
+    private static JDBCDetails POSTGRESQL_DETAILS =
+            new JDBCDetails("postgresql",
+                            "bytea",
+                            "bytea",
+                            "bigint",
+                            true);
+
+    private static JDBCDetails MYSQL_DETAILS =
+            new JDBCDetails("mysql",
+                            "blob",
+                            "varbinary(%d)",
+                            "bigint",
+                            false);
+
+
+    private static JDBCDetails SYBASE_DETAILS =
+            new JDBCDetails("sybase",
+                            "image",
+                            "varbinary(%d)",
+                            "bigint",
+                            false);
+
+
+    private static JDBCDetails ORACLE_DETAILS =
+            new JDBCDetails("oracle",
+                            "blob",
+                            "raw(%d)",
+                            "number",
+                            false);
+
+
+    private static Map<String, JDBCDetails> VENDOR_DETAILS = new HashMap<String,JDBCDetails>();
+
+    static
+    {
+
+        addDetails(DERBY_DETAILS);
+        addDetails(POSTGRESQL_DETAILS);
+        addDetails(MYSQL_DETAILS);
+        addDetails(SYBASE_DETAILS);
+        addDetails(ORACLE_DETAILS);
+    }
+
+    private static void addDetails(JDBCDetails details)
+    {
+        VENDOR_DETAILS.put(details.getVendor(), details);
+    }
+
+    private String _blobType;
+    private String _varBinaryType;
+    private String _bigIntType;
+    private boolean _useBytesMethodsForBlob;
+
+    private List<RecordedJDBCTransaction> _transactions = new CopyOnWriteArrayList<RecordedJDBCTransaction>();
+
+
+    public JDBCMessageStore()
+    {
+    }
+
+    protected Logger getLogger()
+    {
+        return _logger;
+    }
+
+    protected String getSqlBlobType()
+    {
+        return _blobType;
+    }
+
+    protected String getSqlVarBinaryType(int size)
+    {
+        return String.format(_varBinaryType, size);
+    }
+
+    public String getSqlBigIntType()
+    {
+        return _bigIntType;
+    }
+
+    @Override
+    protected void doClose() throws AMQStoreException
+    {
+        while(!_transactions.isEmpty())
+        {
+            RecordedJDBCTransaction txn = _transactions.get(0);
+            txn.abortTran();
+        }
+        try
+        {
+            _connectionProvider.close();
+        }
+        catch (SQLException e)
+        {
+            throw new AMQStoreException("Unable to close connection provider ", e);
+        }
+    }
+
+
+    protected Connection getConnection() throws SQLException
+    {
+        return _connectionProvider.getConnection();
+    }
+
+
+    protected void implementationSpecificConfiguration(String name,
+                                                       VirtualHost virtualHost)
+        throws ClassNotFoundException, SQLException
+    {
+
+
+        String connectionURL = virtualHost.getAttribute(CONNECTION_URL) == null
+                               ? String.valueOf(virtualHost.getAttribute(VirtualHost.STORE_PATH))
+                               : String.valueOf(virtualHost.getAttribute(CONNECTION_URL));
+
+        JDBCDetails details = null;
+
+        String[] components = connectionURL.split(":",3);
+        if(components.length >= 2)
+        {
+            String vendor = components[1];
+            details = VENDOR_DETAILS.get(vendor);
+        }
+
+        if(details == null)
+        {
+            getLogger().info("Do not recognize vendor from connection URL: " + connectionURL);
+
+            // TODO - is there a better default than derby
+            details = DERBY_DETAILS;
+        }
+
+
+        Object poolAttribute = virtualHost.getAttribute("connectionPool");
+        String connectionPoolType = poolAttribute == null ? "DEFAULT" : String.valueOf(poolAttribute);
+
+        JDBCConnectionProviderFactory connectionProviderFactory =
+                JDBCConnectionProviderFactory.FACTORIES.get(connectionPoolType);
+        if(connectionProviderFactory == null)
+        {
+            _logger.warn("Unknown connection pool type: " + connectionPoolType + ".  no connection pooling will be used");
+            connectionProviderFactory = new DefaultConnectionProviderFactory();
+        }
+
+        _connectionProvider = connectionProviderFactory.getConnectionProvider(connectionURL, virtualHost);
+
+        _blobType = getStringAttribute(virtualHost, "jdbcBlobType",details.getBlobType());
+        _varBinaryType = getStringAttribute(virtualHost, "jdbcVarbinaryType",details.getVarBinaryType());
+        _useBytesMethodsForBlob = getBooleanAttribute(virtualHost, "jdbcBytesForBlob",details.isUseBytesMethodsForBlob());
+        _bigIntType = getStringAttribute(virtualHost, "jdbcBigIntType", details.getBigintType());
+    }
+
+
+    private String getStringAttribute(VirtualHost virtualHost, String attributeName, String defaultVal)
+    {
+        Object attrValue = virtualHost.getAttribute(attributeName);
+        if(attrValue != null)
+        {
+            return attrValue.toString();
+        }
+        return defaultVal;
+    }
+
+    private boolean getBooleanAttribute(VirtualHost virtualHost, String attributeName, boolean defaultVal)
+    {
+        Object attrValue = virtualHost.getAttribute(attributeName);
+        if(attrValue != null)
+        {
+            if(attrValue instanceof Boolean)
+            {
+                return ((Boolean) attrValue).booleanValue();
+            }
+            else if(attrValue instanceof String)
+            {
+                return Boolean.parseBoolean((String)attrValue);
+            }
+
+        }
+        return defaultVal;
+    }
+
+
+    protected void storedSizeChange(int contentSize)
+    {
+    }
+
+    @Override
+    public String getStoreLocation()
+    {
+        return "";
+    }
+
+    @Override
+    public String getStoreType()
+    {
+        return TYPE;
+    }
+
+    @Override
+    protected byte[] getBlobAsBytes(ResultSet rs, int col) throws SQLException
+    {
+        if(_useBytesMethodsForBlob)
+        {
+            return rs.getBytes(col);
+        }
+        else
+        {
+            Blob dataAsBlob = rs.getBlob(col);
+            return dataAsBlob.getBytes(1,(int) dataAsBlob.length());
+
+        }
+    }
+
+    @Override
+    protected String getBlobAsString(ResultSet rs, int col) throws SQLException
+    {
+        byte[] bytes;
+        if(_useBytesMethodsForBlob)
+        {
+            bytes = rs.getBytes(col);
+            return new String(bytes,UTF8_CHARSET);
+        }
+        else
+        {
+            Blob blob = rs.getBlob(col);
+            if(blob == null)
+            {
+                return null;
+            }
+            bytes = blob.getBytes(1, (int)blob.length());
+        }
+        return new String(bytes, UTF8_CHARSET);
+
+    }
+
+    @Override
+    public Transaction newTransaction()
+    {
+        return new RecordedJDBCTransaction();
+    }
+
+    private class RecordedJDBCTransaction extends JDBCTransaction
+    {
+        private RecordedJDBCTransaction()
+        {
+            super();
+            JDBCMessageStore.this._transactions.add(this);
+        }
+
+        @Override
+        public void commitTran() throws AMQStoreException
+        {
+            try
+            {
+                super.commitTran();
+            }
+            finally
+            {
+                JDBCMessageStore.this._transactions.remove(this);
+            }
+        }
+
+        @Override
+        public StoreFuture commitTranAsync() throws AMQStoreException
+        {
+            try
+            {
+                return super.commitTranAsync();
+            }
+            finally
+            {
+                JDBCMessageStore.this._transactions.remove(this);
+            }
+        }
+
+        @Override
+        public void abortTran() throws AMQStoreException
+        {
+            try
+            {
+                super.abortTran();
+            }
+            finally
+            {
+                JDBCMessageStore.this._transactions.remove(this);
+            }
+        }
+    }
+
+}
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreFactory.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreFactory.java
new file mode 100644
index 0000000000..82d2275156
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreFactory.java
@@ -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.jdbc;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.configuration.Configuration;
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.plugin.MessageStoreFactory;
+import org.apache.qpid.server.store.MessageStore;
+
+public class JDBCMessageStoreFactory implements MessageStoreFactory
+{
+
+    @Override
+    public String getType()
+    {
+        return JDBCMessageStore.TYPE;
+    }
+
+    @Override
+    public MessageStore createMessageStore()
+    {
+        return new JDBCMessageStore();
+    }
+
+    @Override
+    public Map<String, Object> convertStoreConfiguration(Configuration storeConfiguration)
+    {
+        Map<String,Object> convertedMap = new HashMap<String,Object>();
+        convertedMap.put("jdbcBlobType", storeConfiguration.getString("sqlBlobType"));
+        convertedMap.put("jdbcVarbinaryType", storeConfiguration.getString("sqlVarbinaryType"));
+        if(storeConfiguration.containsKey("useBytesForBlob"))
+        {
+            convertedMap.put("jdbcUseBytesForBlob", storeConfiguration.getBoolean("useBytesForBlob"));
+        }
+        convertedMap.put("jdbcBigIntType", storeConfiguration.getString("sqlBigIntType"));
+        convertedMap.put("connectionPool", storeConfiguration.getString("pool.type"));
+        convertedMap.put("minConnectionsPerPartition", storeConfiguration.getInteger("pool.minConnectionsPerPartition",
+                null));
+        convertedMap.put("maxConnectionsPerPartition", storeConfiguration.getInteger("pool.maxConnectionsPerPartition",
+                null));
+        convertedMap.put("partitionCount", storeConfiguration.getInteger("pool.partitionCount", null));
+
+        return convertedMap;
+    }
+
+
+    @Override
+    public void validateAttributes(Map<String, Object> attributes)
+    {
+        Object connectionURL = attributes.get(JDBCMessageStore.CONNECTION_URL);
+        if(!(connectionURL instanceof String))
+        {
+            Object storePath = attributes.get(VirtualHost.STORE_PATH);
+            if(!(storePath instanceof String))
+            {
+                throw new IllegalArgumentException("Attribute '"+ JDBCMessageStore.CONNECTION_URL
+                                                               +"' is required and must be of type String.");
+
+            }
+        }
+    }
+
+}
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/js/qpid/management/virtualhost/store/jdbc/add.js b/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/js/qpid/management/virtualhost/store/jdbc/add.js
new file mode 100644
index 0000000000..dd79aae2fa
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/js/qpid/management/virtualhost/store/jdbc/add.js
@@ -0,0 +1,93 @@
+/*
+ *
+ * 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.
+ *
+ */
+define(["dojo/_base/xhr",
+        "dojo/dom",
+        "dojo/dom-construct",
+        "dojo/_base/window",
+        "dijit/registry",
+        "dojo/parser",
+        "dojo/_base/array",
+        "dojo/_base/event",
+        "dojo/_base/json",
+        "dojo/string",
+        "dojo/store/Memory",
+        "dijit/form/FilteringSelect",
+        "dojo/domReady!"],
+    function (xhr, dom, construct, win, registry, parser, array, event, json, string, Memory, FilteringSelect) {
+        return {
+            show: function() {
+                var node = dom.byId("addVirtualHost.storeSpecificDiv");
+                var that = this;
+
+                array.forEach(registry.toArray(),
+                              function(item) {
+                                  if(item.id.substr(0,34) == "formAddVirtualHost.specific.store.") {
+                                      item.destroyRecursive();
+                                  }
+                              });
+
+                xhr.get({url: "virtualhost/store/jdbc/add.html",
+                     sync: true,
+                     load:  function(data) {
+                                 node.innerHTML = data;
+                                 parser.parse(node);
+
+                                 if (that.hasOwnProperty("poolTypeChooser"))
+                                 {
+                                     that.poolTypeChooser.destroy();
+                                 }
+
+                                 var selectPoolType = function(type) {
+                                     if(type && string.trim(type) != "") {
+                                         require(["qpid/management/virtualhost/store/pool/"+type.toLowerCase()+"/add"],
+                                         function(poolType)
+                                         {
+                                             poolType.show();
+                                         });
+                                     }
+                                 }
+
+                                 xhr.get({
+                                     sync: true,
+                                     url: "rest/helper?action=pluginList&plugin=JDBCConnectionProviderFactory",
+                                     handleAs: "json"
+                                 }).then(
+                                     function(data) {
+                                         var poolTypes =  data;
+                                         var poolTypesData = [];
+                                         for (var i =0 ; i < poolTypes.length; i++)
+                                         {
+                                             poolTypesData[i]= {id: poolTypes[i], name: poolTypes[i]};
+                                         }
+                                         var poolTypesStore = new Memory({ data: poolTypesData });
+                                         var poolTypesDiv = dom.byId("addVirtualHost.specific.selectPoolType");
+                                         var input = construct.create("input", {id: "addPoolType", required: false}, poolTypesDiv);
+                                         that.poolTypeChooser = new FilteringSelect({ id: "addVirtualHost.specific.store.poolType",
+                                                                                   name: "connectionPool",
+                                                                                   store: poolTypesStore,
+                                                                                   searchAttr: "name", required: false,
+                                                                                   onChange: selectPoolType }, input);
+                                 });
+
+                     }});
+            }
+        };
+    });
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/js/qpid/management/virtualhost/store/pool/none/add.js b/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/js/qpid/management/virtualhost/store/pool/none/add.js
new file mode 100644
index 0000000000..7276737873
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/js/qpid/management/virtualhost/store/pool/none/add.js
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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.
+ *
+ */
+define(["dojo/_base/xhr",
+        "dojo/dom",
+        "dojo/dom-construct",
+        "dojo/_base/window",
+        "dijit/registry",
+        "dojo/parser",
+        "dojo/_base/array",
+        "dojo/_base/event",
+        "dojo/_base/json",
+        "dojo/string",
+        "dojo/store/Memory",
+        "dijit/form/FilteringSelect",
+        "dojo/domReady!"],
+    function (xhr, dom, construct, win, registry, parser, array, event, json, string, Memory, FilteringSelect) {
+        return {
+            show: function() {
+                var node = dom.byId("addVirtualHost.poolSpecificDiv");
+                var that = this;
+
+                array.forEach(registry.toArray(),
+                              function(item) {
+                                  if(item.id.substr(0,39) == "formAddVirtualHost.specific.store.pool.") {
+                                      item.destroyRecursive();
+                                  }
+                              });
+
+                xhr.get({url: "virtualhost/store/pool/none/add.html",
+                     sync: true,
+                     load:  function(data) {
+                                node.innerHTML = data;
+                                parser.parse(node);
+
+                     }});
+            }
+        };
+    });
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/virtualhost/store/jdbc/add.html b/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/virtualhost/store/jdbc/add.html
new file mode 100644
index 0000000000..966b4fcc06
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/resources/virtualhost/store/jdbc/add.html
@@ -0,0 +1,15 @@
+<table class="tableContainer-table tableContainer-table-horiz">
+  <tr>
+      <td class="tableContainer-labelCell" style="width: 300px;"><strong>JDBC Url*:  </strong></td>
+      <td class="tableContainer-valueCell">
+          <input dojoType="dijit/form/ValidationTextBox" required="true" id="formAddVirtualHost.specific.store.connectionURL"
+                 name="connectionURL" placeholder="jdbc:provider:info" />
+  </tr>
+  <tr>
+      <td class="tableContainer-labelCell" style="width: 300px;"><strong>Connection Pool:  </strong></td>
+      <td class="tableContainer-valueCell" ><div id="addVirtualHost.specific.selectPoolType"></div></td>
+  </td>
+  </tr>
+</table>
+<div id="addVirtualHost.poolSpecificDiv">
+</div>
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/resources/services/org.apache.qpid.server.plugin.MessageStoreFactory b/qpid/java/broker-plugins/jdbc-store/src/main/resources/services/org.apache.qpid.server.plugin.MessageStoreFactory
new file mode 100644
index 0000000000..a77458f27d
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/resources/services/org.apache.qpid.server.plugin.MessageStoreFactory
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+org.apache.qpid.server.store.jdbc.JDBCMessageStoreFactory
diff --git a/qpid/java/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java b/qpid/java/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
new file mode 100644
index 0000000000..9c348383c6
--- /dev/null
+++ b/qpid/java/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
@@ -0,0 +1,152 @@
+/*
+ *
+ * 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.jdbc;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.MessageStoreTestCase;
+
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.when;
+
+public class JDBCMessageStoreTest extends MessageStoreTestCase
+{
+    private String _connectionURL;
+
+    @Override
+    public void tearDown() throws Exception
+    {
+        try
+        {
+            shutdownDerby();
+        }
+        finally
+        {
+            super.tearDown();
+        }
+    }
+
+    public void testOnDelete() throws Exception
+    {
+        String[] expectedTables = JDBCMessageStore.ALL_TABLES;
+        assertTablesExist(expectedTables, true);
+        getStore().close();
+        assertTablesExist(expectedTables, true);
+        getStore().onDelete();
+        assertTablesExist(expectedTables, false);
+    }
+
+    @Override
+    protected void setUpStoreConfiguration(VirtualHost virtualHost) throws Exception
+    {
+        _connectionURL = "jdbc:derby:memory:/" + getTestName() + ";create=true";
+
+        when(virtualHost.getAttribute(eq("connectionURL"))).thenReturn(_connectionURL);
+    }
+
+    @Override
+    protected MessageStore createMessageStore()
+    {
+        return new JDBCMessageStore();
+    }
+
+    private void assertTablesExist(String[] expectedTables, boolean exists) throws SQLException
+    {
+        Set<String> existingTables = getTableNames();
+        for (String tableName : expectedTables)
+        {
+            assertEquals("Table " + tableName + (exists ? " is not found" : " actually exist"), exists,
+                    existingTables.contains(tableName));
+        }
+    }
+
+    private Set<String> getTableNames() throws SQLException
+    {
+        Set<String> tableNames = new HashSet<String>();
+        Connection conn = null;
+        try
+        {
+            conn = openConnection();
+            DatabaseMetaData metaData = conn.getMetaData();
+            ResultSet tables = metaData.getTables(null, null, null, new String[] { "TABLE" });
+            try
+            {
+                while (tables.next())
+                {
+                    tableNames.add(tables.getString("TABLE_NAME"));
+                }
+            }
+            finally
+            {
+                tables.close();
+            }
+        }
+        finally
+        {
+            if (conn != null)
+            {
+                conn.close();
+            }
+        }
+        return tableNames;
+    }
+
+    private Connection openConnection() throws SQLException
+    {
+        return DriverManager.getConnection(_connectionURL);
+    }
+
+
+    private void shutdownDerby() throws SQLException
+    {
+        Connection connection = null;
+        try
+        {
+            connection = DriverManager.getConnection("jdbc:derby:memory:/" + getTestName() + ";shutdown=true");
+        }
+        catch(SQLException e)
+        {
+            if (e.getSQLState().equalsIgnoreCase("08006"))
+            {
+                //expected and represents a clean shutdown of this database only, do nothing.
+            }
+            else
+            {
+                throw e;
+            }
+        }
+        finally
+        {
+            if (connection != null)
+            {
+                connection.close();
+            }
+        }
+    }
+}