save master slave rule config to zk

sharding-jdbc-core/src/main/java/com/dangdang/ddframe/rdb/sharding/jdbc/core/datasource/MasterSlaveDataSource.java

@@ -17,11 +17,11 @@
 
 package com.dangdang.ddframe.rdb.sharding.jdbc.core.datasource;
 
-import com.dangdang.ddframe.rdb.sharding.rule.MasterSlaveRule;
 import com.dangdang.ddframe.rdb.sharding.constant.SQLType;
 import com.dangdang.ddframe.rdb.sharding.hint.HintManagerHolder;
 import com.dangdang.ddframe.rdb.sharding.jdbc.adapter.AbstractDataSourceAdapter;
 import com.dangdang.ddframe.rdb.sharding.jdbc.core.connection.MasterSlaveConnection;
+import com.dangdang.ddframe.rdb.sharding.rule.MasterSlaveRule;
 import com.google.common.base.Preconditions;
 import lombok.Getter;
 
@@ -30,6 +30,7 @@ import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.Map;
@@ -50,7 +51,7 @@ public class MasterSlaveDataSource extends AbstractDataSourceAdapter {
         }
     };
     
-    private final MasterSlaveRule masterSlaveRule;
+    private MasterSlaveRule masterSlaveRule;
     
     public MasterSlaveDataSource(final MasterSlaveRule masterSlaveRule) throws SQLException {
         super(getAllDataSources(masterSlaveRule.getMasterDataSource(), masterSlaveRule.getSlaveDataSourceMap().values()));
@@ -105,6 +106,18 @@ public class MasterSlaveDataSource extends AbstractDataSourceAdapter {
         return SQLType.DQL != sqlType || DML_FLAG.get() || HintManagerHolder.isMasterRouteOnly();
     }
     
+    /**
+     * Renew master-slave data source.
+     *
+     * @param masterSlaveRule new master-slave rule
+     * @throws SQLException SQL exception
+     */
+    public void renew(final MasterSlaveRule masterSlaveRule) throws SQLException {
+        Preconditions.checkState(getDatabaseType() == getDatabaseType(Collections.singletonList(masterSlaveRule.getMasterDataSource())), "Cannot change database type dynamically.");
+        Preconditions.checkState(getDatabaseType() == getDatabaseType(masterSlaveRule.getSlaveDataSourceMap().values()), "Cannot change database type dynamically.");
+        this.masterSlaveRule = masterSlaveRule;
+    }
+    
     @Override
     public Connection getConnection() throws SQLException {
         return new MasterSlaveConnection(this);

sharding-jdbc-orchestration/src/main/java/com/dangdang/ddframe/rdb/sharding/api/OrchestrationMasterSlaveDataSourceFactory.java

+/*
+ * Copyright 1999-2015 dangdang.com.
+ * <p>
+ * Licensed 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.
+ * </p>
+ */
+
+package com.dangdang.ddframe.rdb.sharding.api;
+
+import com.dangdang.ddframe.rdb.sharding.api.config.MasterSlaveRuleConfig;
+import com.dangdang.ddframe.rdb.sharding.jdbc.core.datasource.MasterSlaveDataSource;
+import com.dangdang.ddframe.rdb.sharding.json.DataSourceJsonConverter;
+import com.dangdang.ddframe.rdb.sharding.json.GsonFactory;
+import com.dangdang.ddframe.rdb.sharding.reg.base.CoordinatorRegistryCenter;
+import com.google.common.base.Charsets;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.TreeCache;
+import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
+import org.apache.curator.framework.recipes.cache.TreeCacheListener;
+
+import javax.sql.DataSource;
+import java.sql.SQLException;
+import java.util.Map;
+
+/**
+ * Orchestration master slave data source factory.
+ * 
+ * @author zhangliang 
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class OrchestrationMasterSlaveDataSourceFactory {
+    
+    /**
+     * Create sharding data source.
+     * 
+     * @param name name of sharding data source
+     * @param registryCenter registry center
+     * @param dataSourceMap data source map
+     * @param masterSlaveRuleConfig rule configuration for databases and tables sharding
+     * @return sharding data source
+     * @throws SQLException SQL exception
+     */
+    public static DataSource createDataSource(
+            final String name, final CoordinatorRegistryCenter registryCenter, final Map<String, DataSource> dataSourceMap, final MasterSlaveRuleConfig masterSlaveRuleConfig) throws SQLException {
+        initRegistryCenter(name, registryCenter, dataSourceMap, masterSlaveRuleConfig);
+        MasterSlaveDataSource result = (MasterSlaveDataSource) MasterSlaveDataSourceFactory.createDataSource(dataSourceMap, masterSlaveRuleConfig);
+        addConfigurationChangeListener(name, registryCenter, result);
+        return result;
+    }
+    
+    private static void initRegistryCenter(final String name, 
+                                           final CoordinatorRegistryCenter registryCenter, final Map<String, DataSource> dataSourceMap, final MasterSlaveRuleConfig masterSlaveRuleConfig) {
+        registryCenter.init();
+        registryCenter.persist("/" + name + "/config/datasource", DataSourceJsonConverter.toJson(dataSourceMap));
+        registryCenter.persist("/" + name + "/config/masterslave", GsonFactory.getGson().toJson(masterSlaveRuleConfig));
+        registryCenter.addCacheData("/" + name + "/config");
+    }
+    
+    private static void addConfigurationChangeListener(final String name, final CoordinatorRegistryCenter registryCenter, final MasterSlaveDataSource masterSlaveDataSource) {
+        TreeCache cache = (TreeCache) registryCenter.getRawCache("/" + name + "/config");
+        cache.getListenable().addListener(new TreeCacheListener() {
+            
+            @Override
+            public void childEvent(final CuratorFramework client, final TreeCacheEvent event) throws Exception {
+                ChildData childData = event.getData();
+                if (null == childData || null == childData.getData()) {
+                    return;
+                }
+                String path = childData.getPath();
+                if (path.isEmpty()) {
+                    return;
+                }
+                if (("/" + name + "/config/datasource").equals(path)) {
+                    Map<String, DataSource> newDataSourceMap = DataSourceJsonConverter.fromJson(new String(childData.getData(), Charsets.UTF_8));
+                    MasterSlaveRuleConfig masterSlaveRuleConfig = GsonFactory.getGson().fromJson(registryCenter.get("/" + name + "/config/masterslave"), MasterSlaveRuleConfig.class);
+                    masterSlaveDataSource.renew(masterSlaveRuleConfig.build(newDataSourceMap));
+                } else if (("/" + name + "/config/masterslave").equals(path)) {
+                    MasterSlaveRuleConfig newMasterSlaveRuleConfig = GsonFactory.getGson().fromJson(new String(childData.getData(), Charsets.UTF_8), MasterSlaveRuleConfig.class);
+                    Map<String, DataSource> dataSourceMap = DataSourceJsonConverter.fromJson(registryCenter.get("/" + name + "/config/datasource"));
+                    masterSlaveDataSource.renew(newMasterSlaveRuleConfig.build(dataSourceMap));
+                }
+            }
+        });
+    }
+}

sharding-jdbc-orchestration/src/main/java/com/dangdang/ddframe/rdb/sharding/api/OrchestrationShardingDataSourceFactory.java

@@ -57,7 +57,7 @@ public final class OrchestrationShardingDataSourceFactory {
     public static DataSource createDataSource(
             final String name, final CoordinatorRegistryCenter registryCenter, final Map<String, DataSource> dataSourceMap, final ShardingRuleConfig shardingRuleConfig) throws SQLException {
         initRegistryCenter(name, registryCenter, dataSourceMap, shardingRuleConfig);
-        ShardingDataSource result = new ShardingDataSource(shardingRuleConfig.build(dataSourceMap));
+        ShardingDataSource result = (ShardingDataSource) ShardingDataSourceFactory.createDataSource(dataSourceMap, shardingRuleConfig);
         addConfigurationChangeListener(name, registryCenter, result);
         return result;
     }
@@ -78,7 +78,7 @@ public final class OrchestrationShardingDataSourceFactory {
             final ShardingRuleConfig shardingRuleConfig, final Properties props) throws SQLException {
         initRegistryCenter(name, registryCenter, dataSourceMap, shardingRuleConfig);
         // TODO props
-        ShardingDataSource result = new ShardingDataSource(shardingRuleConfig.build(dataSourceMap), props);
+        ShardingDataSource result = (ShardingDataSource) ShardingDataSourceFactory.createDataSource(dataSourceMap, shardingRuleConfig, props);
         addConfigurationChangeListener(name, registryCenter, result);
         return result;
     }