|
1 | 1 | /* |
2 | | - * Copyright 2012-2019 the original author or authors. |
| 2 | + * Copyright 2012-2020 the original author or authors. |
3 | 3 | * |
4 | 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
5 | 5 | * you may not use this file except in compliance with the License. |
|
16 | 16 |
|
17 | 17 | package org.springframework.boot.autoconfigure.cassandra; |
18 | 18 |
|
| 19 | +import java.security.NoSuchAlgorithmException; |
19 | 20 | import java.time.Duration; |
| 21 | +import java.util.LinkedHashMap; |
| 22 | +import java.util.List; |
| 23 | +import java.util.Map; |
20 | 24 |
|
21 | | -import com.datastax.driver.core.Cluster; |
22 | | -import com.datastax.driver.core.PoolingOptions; |
23 | | -import com.datastax.driver.core.QueryOptions; |
24 | | -import com.datastax.driver.core.SocketOptions; |
| 25 | +import javax.net.ssl.SSLContext; |
| 26 | + |
| 27 | +import com.datastax.oss.driver.api.core.CqlSession; |
| 28 | +import com.datastax.oss.driver.api.core.CqlSessionBuilder; |
| 29 | +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; |
| 30 | +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; |
| 31 | +import com.datastax.oss.driver.api.core.config.DriverOption; |
| 32 | +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; |
| 33 | +import com.datastax.oss.driver.internal.core.config.typesafe.DefaultDriverConfigLoader; |
| 34 | +import com.datastax.oss.driver.internal.core.config.typesafe.DefaultProgrammaticDriverConfigLoaderBuilder; |
| 35 | +import com.typesafe.config.Config; |
| 36 | +import com.typesafe.config.ConfigFactory; |
25 | 37 |
|
26 | 38 | import org.springframework.beans.factory.ObjectProvider; |
27 | 39 | import org.springframework.boot.autoconfigure.EnableAutoConfiguration; |
|
31 | 43 | import org.springframework.boot.context.properties.PropertyMapper; |
32 | 44 | import org.springframework.context.annotation.Bean; |
33 | 45 | import org.springframework.context.annotation.Configuration; |
34 | | -import org.springframework.util.StringUtils; |
| 46 | +import org.springframework.context.annotation.Lazy; |
35 | 47 |
|
36 | 48 | /** |
37 | 49 | * {@link EnableAutoConfiguration Auto-configuration} for Cassandra. |
|
44 | 56 | * @since 1.3.0 |
45 | 57 | */ |
46 | 58 | @Configuration(proxyBeanMethods = false) |
47 | | -@ConditionalOnClass({ Cluster.class }) |
| 59 | +@ConditionalOnClass({ CqlSession.class }) |
48 | 60 | @EnableConfigurationProperties(CassandraProperties.class) |
49 | 61 | public class CassandraAutoConfiguration { |
50 | 62 |
|
51 | 63 | @Bean |
52 | 64 | @ConditionalOnMissingBean |
53 | | - public Cluster cassandraCluster(CassandraProperties properties, |
54 | | - ObjectProvider<ClusterBuilderCustomizer> builderCustomizers, |
55 | | - ObjectProvider<ClusterFactory> clusterFactory) { |
| 65 | + @Lazy |
| 66 | + public CqlSession cqlSession(CqlSessionBuilder cqlSessionBuilder) { |
| 67 | + return cqlSessionBuilder.build(); |
| 68 | + } |
| 69 | + |
| 70 | + @Bean |
| 71 | + @ConditionalOnMissingBean |
| 72 | + public CqlSessionBuilder cqlSessionBuilder(CassandraProperties properties, DriverConfigLoader driverConfigLoader, |
| 73 | + ObjectProvider<CqlSessionBuilderCustomizer> builderCustomizers) { |
| 74 | + CqlSessionBuilder builder = CqlSession.builder().withConfigLoader(driverConfigLoader); |
| 75 | + configureSsl(properties, builder); |
| 76 | + builder.withKeyspace(properties.getKeyspaceName()); |
| 77 | + builderCustomizers.orderedStream().forEach((customizer) -> customizer.customize(builder)); |
| 78 | + return builder; |
| 79 | + } |
| 80 | + |
| 81 | + private void configureSsl(CassandraProperties properties, CqlSessionBuilder builder) { |
| 82 | + if (properties.isSsl()) { |
| 83 | + try { |
| 84 | + builder.withSslContext(SSLContext.getDefault()); |
| 85 | + } |
| 86 | + catch (NoSuchAlgorithmException ex) { |
| 87 | + throw new IllegalStateException("Could not setup SSL default context for Cassandra", ex); |
| 88 | + } |
| 89 | + } |
| 90 | + } |
| 91 | + |
| 92 | + @Bean |
| 93 | + @ConditionalOnMissingBean |
| 94 | + public DriverConfigLoader driverConfigLoader(CassandraProperties properties, |
| 95 | + ObjectProvider<DriverConfigLoaderBuilderCustomizer> builderCustomizers) { |
| 96 | + ProgrammaticDriverConfigLoaderBuilder builder = new DefaultProgrammaticDriverConfigLoaderBuilder( |
| 97 | + () -> cassandraConfiguration(properties), DefaultDriverConfigLoader.DEFAULT_ROOT_PATH); |
| 98 | + builderCustomizers.orderedStream().forEach((customizer) -> customizer.customizer(builder)); |
| 99 | + return builder.build(); |
| 100 | + } |
| 101 | + |
| 102 | + private Config cassandraConfiguration(CassandraProperties properties) { |
| 103 | + CassandraDriverOptions options = new CassandraDriverOptions(); |
56 | 104 | PropertyMapper map = PropertyMapper.get(); |
57 | | - Cluster.Builder builder = Cluster.builder().withClusterName(properties.getClusterName()) |
58 | | - .withPort(properties.getPort()); |
| 105 | + map.from(properties.getSessionName()).whenHasText() |
| 106 | + .to((sessionName) -> options.add(DefaultDriverOption.SESSION_NAME, sessionName)); |
59 | 107 | map.from(properties::getUsername).whenNonNull() |
60 | | - .to((username) -> builder.withCredentials(username, properties.getPassword())); |
61 | | - map.from(properties::getCompression).whenNonNull().to(builder::withCompression); |
62 | | - QueryOptions queryOptions = getQueryOptions(properties); |
63 | | - map.from(queryOptions).to(builder::withQueryOptions); |
64 | | - SocketOptions socketOptions = getSocketOptions(properties); |
65 | | - map.from(socketOptions).to(builder::withSocketOptions); |
66 | | - map.from(properties::isSsl).whenTrue().toCall(builder::withSSL); |
67 | | - PoolingOptions poolingOptions = getPoolingOptions(properties); |
68 | | - map.from(poolingOptions).to(builder::withPoolingOptions); |
69 | | - map.from(properties::getContactPoints).as(StringUtils::toStringArray).to(builder::addContactPoints); |
70 | | - map.from(properties::isJmxEnabled).whenFalse().toCall(builder::withoutJMXReporting); |
71 | | - builderCustomizers.orderedStream().forEach((customizer) -> customizer.customize(builder)); |
72 | | - return clusterFactory.getIfAvailable(() -> Cluster::buildFrom).create(builder); |
| 108 | + .to((username) -> options.add(DefaultDriverOption.AUTH_PROVIDER_USER_NAME, username) |
| 109 | + .add(DefaultDriverOption.AUTH_PROVIDER_PASSWORD, properties.getPassword())); |
| 110 | + map.from(properties::getCompression).whenNonNull() |
| 111 | + .to((compression) -> options.add(DefaultDriverOption.PROTOCOL_COMPRESSION, compression)); |
| 112 | + mapQueryOptions(properties, options); |
| 113 | + mapSocketOptions(properties, options); |
| 114 | + mapPoolingOptions(properties, options); |
| 115 | + map.from(properties::getContactPoints) |
| 116 | + .to((contactPoints) -> options.add(DefaultDriverOption.CONTACT_POINTS, contactPoints)); |
| 117 | + ConfigFactory.invalidateCaches(); |
| 118 | + return ConfigFactory.defaultOverrides().withFallback(options.build()) |
| 119 | + .withFallback(ConfigFactory.defaultReference()).resolve(); |
73 | 120 | } |
74 | 121 |
|
75 | | - private QueryOptions getQueryOptions(CassandraProperties properties) { |
| 122 | + private void mapQueryOptions(CassandraProperties properties, CassandraDriverOptions options) { |
76 | 123 | PropertyMapper map = PropertyMapper.get(); |
77 | | - QueryOptions options = new QueryOptions(); |
78 | | - map.from(properties::getConsistencyLevel).whenNonNull().to(options::setConsistencyLevel); |
79 | | - map.from(properties::getSerialConsistencyLevel).whenNonNull().to(options::setSerialConsistencyLevel); |
80 | | - map.from(properties::getFetchSize).to(options::setFetchSize); |
81 | | - return options; |
| 124 | + map.from(properties::getConsistencyLevel).whenNonNull() |
| 125 | + .to(((consistency) -> options.add(DefaultDriverOption.REQUEST_CONSISTENCY, consistency))); |
| 126 | + map.from(properties::getSerialConsistencyLevel).whenNonNull().to( |
| 127 | + (serialConsistency) -> options.add(DefaultDriverOption.REQUEST_SERIAL_CONSISTENCY, serialConsistency)); |
| 128 | + map.from(properties::getPageSize) |
| 129 | + .to((pageSize) -> options.add(DefaultDriverOption.REQUEST_PAGE_SIZE, pageSize)); |
82 | 130 | } |
83 | 131 |
|
84 | | - private SocketOptions getSocketOptions(CassandraProperties properties) { |
| 132 | + private void mapSocketOptions(CassandraProperties properties, CassandraDriverOptions options) { |
85 | 133 | PropertyMapper map = PropertyMapper.get(); |
86 | | - SocketOptions options = new SocketOptions(); |
87 | 134 | map.from(properties::getConnectTimeout).whenNonNull().asInt(Duration::toMillis) |
88 | | - .to(options::setConnectTimeoutMillis); |
89 | | - map.from(properties::getReadTimeout).whenNonNull().asInt(Duration::toMillis).to(options::setReadTimeoutMillis); |
90 | | - return options; |
| 135 | + .to((connectTimeout) -> options.add(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT, connectTimeout)); |
| 136 | + map.from(properties::getReadTimeout).whenNonNull().asInt(Duration::toMillis) |
| 137 | + .to((readTimeout) -> options.add(DefaultDriverOption.REQUEST_TIMEOUT, readTimeout)); |
91 | 138 | } |
92 | 139 |
|
93 | | - private PoolingOptions getPoolingOptions(CassandraProperties properties) { |
| 140 | + private void mapPoolingOptions(CassandraProperties properties, CassandraDriverOptions options) { |
94 | 141 | PropertyMapper map = PropertyMapper.get(); |
95 | 142 | CassandraProperties.Pool poolProperties = properties.getPool(); |
96 | | - PoolingOptions options = new PoolingOptions(); |
97 | 143 | map.from(poolProperties::getIdleTimeout).whenNonNull().asInt(Duration::getSeconds) |
98 | | - .to(options::setIdleTimeoutSeconds); |
99 | | - map.from(poolProperties::getPoolTimeout).whenNonNull().asInt(Duration::toMillis) |
100 | | - .to(options::setPoolTimeoutMillis); |
| 144 | + .to((idleTimeout) -> options.add(DefaultDriverOption.HEARTBEAT_TIMEOUT, idleTimeout)); |
101 | 145 | map.from(poolProperties::getHeartbeatInterval).whenNonNull().asInt(Duration::getSeconds) |
102 | | - .to(options::setHeartbeatIntervalSeconds); |
103 | | - map.from(poolProperties::getMaxQueueSize).to(options::setMaxQueueSize); |
104 | | - return options; |
| 146 | + .to((heartBeatInterval) -> options.add(DefaultDriverOption.HEARTBEAT_INTERVAL, heartBeatInterval)); |
| 147 | + map.from(poolProperties::getMaxQueueSize) |
| 148 | + .to((maxQueueSize) -> options.add(DefaultDriverOption.REQUEST_THROTTLER_MAX_QUEUE_SIZE, maxQueueSize)); |
| 149 | + } |
| 150 | + |
| 151 | + private static class CassandraDriverOptions { |
| 152 | + |
| 153 | + private final Map<String, String> options = new LinkedHashMap<>(); |
| 154 | + |
| 155 | + private CassandraDriverOptions add(DriverOption option, String value) { |
| 156 | + String key = createKeyFor(option); |
| 157 | + this.options.put(key, value); |
| 158 | + return this; |
| 159 | + } |
| 160 | + |
| 161 | + private CassandraDriverOptions add(DriverOption option, int value) { |
| 162 | + return add(option, String.valueOf(value)); |
| 163 | + } |
| 164 | + |
| 165 | + private CassandraDriverOptions add(DriverOption option, Enum<?> value) { |
| 166 | + return add(option, value.name()); |
| 167 | + } |
| 168 | + |
| 169 | + private CassandraDriverOptions add(DriverOption option, List<String> values) { |
| 170 | + for (int i = 0; i < values.size(); i++) { |
| 171 | + this.options.put(String.format("%s.%s", createKeyFor(option), i), values.get(i)); |
| 172 | + } |
| 173 | + return this; |
| 174 | + } |
| 175 | + |
| 176 | + private Config build() { |
| 177 | + return ConfigFactory.parseMap(this.options, "Environment"); |
| 178 | + } |
| 179 | + |
| 180 | + private static String createKeyFor(DriverOption option) { |
| 181 | + return String.format("%s.%s", DefaultDriverConfigLoader.DEFAULT_ROOT_PATH, option.getPath()); |
| 182 | + } |
| 183 | + |
105 | 184 | } |
106 | 185 |
|
107 | 186 | } |
0 commit comments