Skip to content

Commit

Permalink
Rename StorageNodeAggregator (#28637)
Browse files Browse the repository at this point in the history
  • Loading branch information
terrymanu authored Oct 4, 2023
1 parent f3dec3c commit 2f77a7f
Show file tree
Hide file tree
Showing 8 changed files with 108 additions and 91 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.shardingsphere.infra.datasource.pool.props.creator.DataSourcePoolPropertiesCreator;
import org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.metadata.database.resource.StorageResource;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeUtils;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeAggregator;
import org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnitNodeMapUtils;

import javax.sql.DataSource;
Expand All @@ -49,7 +49,7 @@ public final class DataSourceProvidedDatabaseConfiguration implements DatabaseCo

public DataSourceProvidedDatabaseConfiguration(final Map<String, DataSource> dataSources, final Collection<RuleConfiguration> ruleConfigs) {
this.ruleConfigurations = ruleConfigs;
storageResource = new StorageResource(StorageNodeUtils.getStorageNodeDataSources(dataSources), StorageUnitNodeMapUtils.fromDataSources(dataSources));
storageResource = new StorageResource(StorageNodeAggregator.aggregateDataSources(dataSources), StorageUnitNodeMapUtils.fromDataSources(dataSources));
dataSourcePoolPropertiesMap = createDataSourcePoolPropertiesMap(dataSources);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNode;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeName;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeUtils;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeAggregator;
import org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnitMetaData;
import org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnitNodeMapUtils;

Expand All @@ -48,7 +48,7 @@ public final class ResourceMetaData {
private final Map<String, StorageUnitMetaData> storageUnitMetaDataMap;

public ResourceMetaData(final Map<String, DataSource> dataSources) {
this.dataSources = StorageNodeUtils.getStorageNodeDataSources(dataSources);
this.dataSources = StorageNodeAggregator.aggregateDataSources(dataSources);
Map<String, StorageNode> storageNodes = StorageUnitNodeMapUtils.fromDataSources(dataSources);
Map<String, DataSourcePoolProperties> dataSourcePoolPropertiesMap = dataSources.entrySet().stream().collect(
Collectors.toMap(Entry::getKey, entry -> DataSourcePoolPropertiesCreator.create(entry.getValue()), (oldValue, currentValue) -> oldValue, LinkedHashMap::new));
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
/*
* 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.shardingsphere.infra.metadata.database.resource.node;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.shardingsphere.infra.database.core.connector.url.JdbcUrl;
import org.apache.shardingsphere.infra.database.core.connector.url.StandardJdbcUrlParser;
import org.apache.shardingsphere.infra.database.core.connector.url.UnrecognizedDatabaseURLException;
import org.apache.shardingsphere.infra.database.core.type.DatabaseTypeFactory;
import org.apache.shardingsphere.infra.database.core.type.DatabaseTypeRegistry;
import org.apache.shardingsphere.infra.datasource.pool.props.domain.DataSourcePoolProperties;

import javax.sql.DataSource;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;

/**
* Storage node aggregator.
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class StorageNodeAggregator {

/**
* Aggregate data source map to storage node name grouped.
*
* @param dataSourceMap storage unit name and data source map
* @return storage node name and data source map
*/
public static Map<StorageNodeName, DataSource> aggregateDataSources(final Map<String, DataSource> dataSourceMap) {
return dataSourceMap.entrySet().stream().collect(
Collectors.toMap(entry -> new StorageNodeName(entry.getKey()), Entry::getValue, (oldValue, currentValue) -> currentValue, () -> new LinkedHashMap<>(dataSourceMap.size(), 1F)));
}

/**
* Aggregate data source pool properties map to storage node name grouped.
*
* @param storageUnitDataSourcePoolPropsMap storage unit name and data source pool properties map
* @return storage node name and data source pool properties map
*/
public static Map<StorageNodeName, DataSourcePoolProperties> aggregateDataSourcePoolProperties(final Map<String, DataSourcePoolProperties> storageUnitDataSourcePoolPropsMap) {
Map<StorageNodeName, DataSourcePoolProperties> result = new LinkedHashMap<>();
for (Entry<String, DataSourcePoolProperties> entry : storageUnitDataSourcePoolPropsMap.entrySet()) {
Map<String, Object> standardProps = entry.getValue().getConnectionPropertySynonyms().getStandardProperties();
String url = standardProps.get("url").toString();
boolean isInstanceConnectionAvailable = new DatabaseTypeRegistry(DatabaseTypeFactory.get(url)).getDialectDatabaseMetaData().isInstanceConnectionAvailable();
StorageNodeName storageNodeName = getStorageNodeName(entry.getKey(), url, standardProps.get("username").toString(), isInstanceConnectionAvailable);
result.putIfAbsent(storageNodeName, entry.getValue());
}
return result;
}

private static StorageNodeName getStorageNodeName(final String dataSourceName, final String url, final String username, final boolean isInstanceConnectionAvailable) {
try {
JdbcUrl jdbcUrl = new StandardJdbcUrlParser().parse(url);
return isInstanceConnectionAvailable ? new StorageNodeName(jdbcUrl.getHostname(), jdbcUrl.getPort(), username) : new StorageNodeName(dataSourceName);
} catch (final UnrecognizedDatabaseURLException ex) {
return new StorageNodeName(dataSourceName);
}
}
}

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -93,22 +93,4 @@ private static StorageNodeName getStorageNodeName(final String dataSourceName, f
private static StorageNode createStorageNode(final StorageNodeName storageNodeName, final String url, final boolean isInstanceConnectionAvailable) {
return isInstanceConnectionAvailable ? new StorageNode(storageNodeName, url, new StandardJdbcUrlParser().parse(url).getDatabase()) : new StorageNode(storageNodeName, url);
}

/**
* Get storage node grouped data source pool properties map.
*
* @param storageUnitDataSourcePoolProps storage unit grouped data source pool properties map
* @return storage node grouped data source pool properties map
*/
public static Map<StorageNodeName, DataSourcePoolProperties> getStorageNodeDataSourcePoolProperties(final Map<String, DataSourcePoolProperties> storageUnitDataSourcePoolProps) {
Map<StorageNodeName, DataSourcePoolProperties> result = new LinkedHashMap<>();
for (Entry<String, DataSourcePoolProperties> entry : storageUnitDataSourcePoolProps.entrySet()) {
Map<String, Object> standardProps = entry.getValue().getConnectionPropertySynonyms().getStandardProperties();
String url = standardProps.get("url").toString();
boolean isInstanceConnectionAvailable = new DatabaseTypeRegistry(DatabaseTypeFactory.get(url)).getDialectDatabaseMetaData().isInstanceConnectionAvailable();
StorageNodeName storageNodeName = getStorageNodeName(entry.getKey(), url, standardProps.get("username").toString(), isInstanceConnectionAvailable);
result.putIfAbsent(storageNodeName, entry.getValue());
}
return result;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.shardingsphere.infra.metadata.database.resource.StorageResource;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNode;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeName;
import org.apache.shardingsphere.infra.metadata.database.resource.node.StorageNodeAggregator;
import org.apache.shardingsphere.infra.metadata.database.resource.unit.StorageUnitNodeMapUtils;

import javax.sql.DataSource;
Expand All @@ -43,26 +44,26 @@ public final class NewResourceSwitchManager {
* Register storage unit.
*
* @param resourceMetaData resource meta data
* @param storageUnitDataSourcePoolProps storage unit grouped data source pool properties map
* @param storageUnitDataSourcePoolPropsMap storage unit grouped data source pool properties map
* @return created switching resource
*/
public SwitchingResource registerStorageUnit(final ResourceMetaData resourceMetaData, final Map<String, DataSourcePoolProperties> storageUnitDataSourcePoolProps) {
public SwitchingResource registerStorageUnit(final ResourceMetaData resourceMetaData, final Map<String, DataSourcePoolProperties> storageUnitDataSourcePoolPropsMap) {
Map<String, DataSourcePoolProperties> mergedPropsMap = new LinkedHashMap<>(resourceMetaData.getStorageUnitMetaDataMap().entrySet().stream()
.collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().getStorageUnit().getDataSourcePoolProperties(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new)));
mergedPropsMap.putAll(storageUnitDataSourcePoolProps);
Map<String, StorageNode> toBeCreatedStorageUintNodeMap = StorageUnitNodeMapUtils.fromDataSourcePoolProperties(storageUnitDataSourcePoolProps);
return new SwitchingResource(resourceMetaData, getRegisterNewStorageResource(resourceMetaData, toBeCreatedStorageUintNodeMap, storageUnitDataSourcePoolProps),
mergedPropsMap.putAll(storageUnitDataSourcePoolPropsMap);
Map<String, StorageNode> toBeCreatedStorageUintNodeMap = StorageUnitNodeMapUtils.fromDataSourcePoolProperties(storageUnitDataSourcePoolPropsMap);
return new SwitchingResource(resourceMetaData,
getRegisterNewStorageResource(resourceMetaData, toBeCreatedStorageUintNodeMap, StorageNodeAggregator.aggregateDataSourcePoolProperties(storageUnitDataSourcePoolPropsMap)),
new StorageResource(Collections.emptyMap(), Collections.emptyMap()), mergedPropsMap);
}

private StorageResource getRegisterNewStorageResource(final ResourceMetaData resourceMetaData,
final Map<String, StorageNode> storageUintNodeMap, final Map<String, DataSourcePoolProperties> storageUnitDataSourcePoolProps) {
final Map<String, StorageNode> storageUintNodeMap, final Map<StorageNodeName, DataSourcePoolProperties> dataSourcePoolPropsMap) {
Collection<StorageNodeName> storageNodeNames = storageUintNodeMap.values().stream().map(StorageNode::getName).collect(Collectors.toSet());
Map<StorageNodeName, DataSourcePoolProperties> storageNodeDataSourcePoolProps = StorageUnitNodeMapUtils.getStorageNodeDataSourcePoolProperties(storageUnitDataSourcePoolProps);
Map<StorageNodeName, DataSource> newStorageNodes = new LinkedHashMap<>(storageNodeNames.size(), 1F);
for (StorageNodeName each : storageNodeNames) {
if (!resourceMetaData.getDataSources().containsKey(each)) {
newStorageNodes.put(each, DataSourcePoolCreator.create(storageNodeDataSourcePoolProps.get(each)));
newStorageNodes.put(each, DataSourcePoolCreator.create(dataSourcePoolPropsMap.get(each)));
}
}
return new StorageResource(newStorageNodes, storageUintNodeMap);
Expand All @@ -80,16 +81,16 @@ public SwitchingResource alterStorageUnit(final ResourceMetaData resourceMetaDat
.collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().getStorageUnit().getDataSourcePoolProperties(), (oldValue, currentValue) -> oldValue, LinkedHashMap::new)));
mergedDataSourcePoolPropertiesMap.putAll(propsMap);
Map<String, StorageNode> toBeAlteredStorageUintNodeMap = StorageUnitNodeMapUtils.fromDataSourcePoolProperties(mergedDataSourcePoolPropertiesMap);
return new SwitchingResource(resourceMetaData, getAlterNewStorageResource(toBeAlteredStorageUintNodeMap, mergedDataSourcePoolPropertiesMap),
return new SwitchingResource(resourceMetaData,
getAlterNewStorageResource(toBeAlteredStorageUintNodeMap, StorageNodeAggregator.aggregateDataSourcePoolProperties(mergedDataSourcePoolPropertiesMap)),
getStaleStorageResource(resourceMetaData, toBeAlteredStorageUintNodeMap), mergedDataSourcePoolPropertiesMap);
}

private StorageResource getAlterNewStorageResource(final Map<String, StorageNode> storageUintNodeMap, final Map<String, DataSourcePoolProperties> storageUnitDataSourcePoolProps) {
private StorageResource getAlterNewStorageResource(final Map<String, StorageNode> storageUintNodeMap, final Map<StorageNodeName, DataSourcePoolProperties> dataSourcePoolPropsMap) {
Collection<StorageNodeName> toBeAlteredStorageNodeNames = storageUintNodeMap.values().stream().map(StorageNode::getName).collect(Collectors.toSet());
Map<StorageNodeName, DataSourcePoolProperties> storageNodeDataSourcePoolProps = StorageUnitNodeMapUtils.getStorageNodeDataSourcePoolProperties(storageUnitDataSourcePoolProps);
Map<StorageNodeName, DataSource> storageNodes = new LinkedHashMap<>(toBeAlteredStorageNodeNames.size(), 1F);
for (StorageNodeName each : toBeAlteredStorageNodeNames) {
storageNodes.put(each, DataSourcePoolCreator.create(storageNodeDataSourcePoolProps.get(each)));
storageNodes.put(each, DataSourcePoolCreator.create(dataSourcePoolPropsMap.get(each)));
}
return new StorageResource(storageNodes, storageUintNodeMap);
}
Expand Down
Loading

0 comments on commit 2f77a7f

Please sign in to comment.