169 changed files with 2541 additions and 1903 deletions
@ -0,0 +1,232 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 The Thingsboard Authors |
||||
|
* |
||||
|
* 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.dao.cassandra; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.ConsistencyLevel; |
||||
|
import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; |
||||
|
import com.datastax.oss.driver.api.core.config.DefaultDriverOption; |
||||
|
import com.datastax.oss.driver.api.core.config.DriverConfigLoader; |
||||
|
import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; |
||||
|
import com.datastax.oss.driver.api.core.metrics.DefaultSessionMetric; |
||||
|
import com.datastax.oss.driver.api.core.metrics.DefaultNodeMetric; |
||||
|
import lombok.Data; |
||||
|
import org.apache.commons.lang3.StringUtils; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.context.annotation.Configuration; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.server.dao.util.NoSqlAnyDao; |
||||
|
|
||||
|
import javax.annotation.PostConstruct; |
||||
|
import java.time.Duration; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.Arrays; |
||||
|
import java.util.Collections; |
||||
|
import java.util.List; |
||||
|
|
||||
|
@Component |
||||
|
@Configuration |
||||
|
@Data |
||||
|
@NoSqlAnyDao |
||||
|
public class CassandraDriverOptions { |
||||
|
|
||||
|
private static final String COMMA = ","; |
||||
|
|
||||
|
@Value("${cassandra.cluster_name}") |
||||
|
private String clusterName; |
||||
|
@Value("${cassandra.url}") |
||||
|
private String url; |
||||
|
|
||||
|
@Value("${cassandra.socket.connect_timeout}") |
||||
|
private int connectTimeoutMillis; |
||||
|
@Value("${cassandra.socket.read_timeout}") |
||||
|
private int readTimeoutMillis; |
||||
|
@Value("${cassandra.socket.keep_alive}") |
||||
|
private Boolean keepAlive; |
||||
|
@Value("${cassandra.socket.reuse_address}") |
||||
|
private Boolean reuseAddress; |
||||
|
@Value("${cassandra.socket.so_linger}") |
||||
|
private Integer soLinger; |
||||
|
@Value("${cassandra.socket.tcp_no_delay}") |
||||
|
private Boolean tcpNoDelay; |
||||
|
@Value("${cassandra.socket.receive_buffer_size}") |
||||
|
private Integer receiveBufferSize; |
||||
|
@Value("${cassandra.socket.send_buffer_size}") |
||||
|
private Integer sendBufferSize; |
||||
|
|
||||
|
@Value("${cassandra.max_requests_per_connection_local:32768}") |
||||
|
private int max_requests_local; |
||||
|
@Value("${cassandra.max_requests_per_connection_remote:32768}") |
||||
|
private int max_requests_remote; |
||||
|
|
||||
|
@Value("${cassandra.query.default_fetch_size}") |
||||
|
private Integer defaultFetchSize; |
||||
|
@Value("${cassandra.query.read_consistency_level}") |
||||
|
private String readConsistencyLevel; |
||||
|
@Value("${cassandra.query.write_consistency_level}") |
||||
|
private String writeConsistencyLevel; |
||||
|
|
||||
|
@Value("${cassandra.compression}") |
||||
|
private String compression; |
||||
|
@Value("${cassandra.ssl}") |
||||
|
private Boolean ssl; |
||||
|
@Value("${cassandra.metrics}") |
||||
|
private Boolean metrics; |
||||
|
|
||||
|
@Value("${cassandra.credentials}") |
||||
|
private Boolean credentials; |
||||
|
@Value("${cassandra.username}") |
||||
|
private String username; |
||||
|
@Value("${cassandra.password}") |
||||
|
private String password; |
||||
|
|
||||
|
@Value("${cassandra.init_timeout_ms}") |
||||
|
private long initTimeout; |
||||
|
@Value("${cassandra.init_retry_interval_ms}") |
||||
|
private long initRetryInterval; |
||||
|
|
||||
|
private DriverConfigLoader loader; |
||||
|
|
||||
|
private ConsistencyLevel defaultReadConsistencyLevel; |
||||
|
private ConsistencyLevel defaultWriteConsistencyLevel; |
||||
|
|
||||
|
@PostConstruct |
||||
|
public void initLoader() { |
||||
|
ProgrammaticDriverConfigLoaderBuilder driverConfigBuilder = |
||||
|
DriverConfigLoader.programmaticBuilder(); |
||||
|
|
||||
|
driverConfigBuilder |
||||
|
.withStringList(DefaultDriverOption.CONTACT_POINTS, getContactPoints(url)) |
||||
|
.withString(DefaultDriverOption.SESSION_NAME, clusterName); |
||||
|
|
||||
|
this.initSocketOptions(driverConfigBuilder); |
||||
|
this.initPoolingOptions(driverConfigBuilder); |
||||
|
this.initQueryOptions(driverConfigBuilder); |
||||
|
|
||||
|
driverConfigBuilder.withString(DefaultDriverOption.PROTOCOL_COMPRESSION, |
||||
|
StringUtils.isEmpty(this.compression) ? "none" : this.compression.toLowerCase()); |
||||
|
|
||||
|
if (this.ssl) { |
||||
|
driverConfigBuilder.withString(DefaultDriverOption.SSL_ENGINE_FACTORY_CLASS, |
||||
|
"DefaultSslEngineFactory"); |
||||
|
} |
||||
|
|
||||
|
if (this.metrics) { |
||||
|
driverConfigBuilder.withStringList(DefaultDriverOption.METRICS_SESSION_ENABLED, |
||||
|
Arrays.asList(DefaultSessionMetric.CONNECTED_NODES.getPath(), |
||||
|
DefaultSessionMetric.CQL_REQUESTS.getPath())); |
||||
|
driverConfigBuilder.withStringList(DefaultDriverOption.METRICS_NODE_ENABLED, |
||||
|
Arrays.asList(DefaultNodeMetric.OPEN_CONNECTIONS.getPath(), |
||||
|
DefaultNodeMetric.IN_FLIGHT.getPath())); |
||||
|
} |
||||
|
|
||||
|
if (this.credentials) { |
||||
|
driverConfigBuilder.withString(DefaultDriverOption.AUTH_PROVIDER_CLASS, |
||||
|
"PlainTextAuthProvider"); |
||||
|
driverConfigBuilder.withString(DefaultDriverOption.AUTH_PROVIDER_USER_NAME, |
||||
|
this.username); |
||||
|
driverConfigBuilder.withString(DefaultDriverOption.AUTH_PROVIDER_PASSWORD, |
||||
|
this.password); |
||||
|
} |
||||
|
|
||||
|
driverConfigBuilder.withBoolean(DefaultDriverOption.RECONNECT_ON_INIT, |
||||
|
true); |
||||
|
driverConfigBuilder.withString(DefaultDriverOption.RECONNECTION_POLICY_CLASS, |
||||
|
"ExponentialReconnectionPolicy"); |
||||
|
driverConfigBuilder.withDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY, |
||||
|
Duration.ofMillis(this.initRetryInterval)); |
||||
|
driverConfigBuilder.withDuration(DefaultDriverOption.RECONNECTION_MAX_DELAY, |
||||
|
Duration.ofMillis(this.initTimeout)); |
||||
|
|
||||
|
this.loader = driverConfigBuilder.build(); |
||||
|
} |
||||
|
|
||||
|
protected ConsistencyLevel getDefaultReadConsistencyLevel() { |
||||
|
if (defaultReadConsistencyLevel == null) { |
||||
|
if (readConsistencyLevel != null) { |
||||
|
defaultReadConsistencyLevel = DefaultConsistencyLevel.valueOf(readConsistencyLevel.toUpperCase()); |
||||
|
} else { |
||||
|
defaultReadConsistencyLevel = DefaultConsistencyLevel.ONE; |
||||
|
} |
||||
|
} |
||||
|
return defaultReadConsistencyLevel; |
||||
|
} |
||||
|
|
||||
|
protected ConsistencyLevel getDefaultWriteConsistencyLevel() { |
||||
|
if (defaultWriteConsistencyLevel == null) { |
||||
|
if (writeConsistencyLevel != null) { |
||||
|
defaultWriteConsistencyLevel = DefaultConsistencyLevel.valueOf(writeConsistencyLevel.toUpperCase()); |
||||
|
} else { |
||||
|
defaultWriteConsistencyLevel = DefaultConsistencyLevel.ONE; |
||||
|
} |
||||
|
} |
||||
|
return defaultWriteConsistencyLevel; |
||||
|
} |
||||
|
|
||||
|
private void initSocketOptions(ProgrammaticDriverConfigLoaderBuilder driverConfigBuilder) { |
||||
|
driverConfigBuilder.withDuration(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT, |
||||
|
Duration.ofMillis(this.connectTimeoutMillis)); |
||||
|
driverConfigBuilder.withDuration(DefaultDriverOption.REQUEST_TIMEOUT, |
||||
|
Duration.ofMillis(this.readTimeoutMillis)); |
||||
|
if (this.keepAlive != null) { |
||||
|
driverConfigBuilder.withBoolean(DefaultDriverOption.SOCKET_KEEP_ALIVE, |
||||
|
this.keepAlive); |
||||
|
} |
||||
|
if (this.reuseAddress != null) { |
||||
|
driverConfigBuilder.withBoolean(DefaultDriverOption.SOCKET_REUSE_ADDRESS, |
||||
|
this.reuseAddress); |
||||
|
} |
||||
|
if (this.soLinger != null) { |
||||
|
driverConfigBuilder.withInt(DefaultDriverOption.SOCKET_LINGER_INTERVAL, |
||||
|
this.soLinger); |
||||
|
} |
||||
|
if (this.tcpNoDelay != null) { |
||||
|
driverConfigBuilder.withBoolean(DefaultDriverOption.SOCKET_TCP_NODELAY, |
||||
|
this.tcpNoDelay); |
||||
|
} |
||||
|
if (this.receiveBufferSize != null) { |
||||
|
driverConfigBuilder.withInt(DefaultDriverOption.SOCKET_RECEIVE_BUFFER_SIZE, |
||||
|
this.receiveBufferSize); |
||||
|
} |
||||
|
if (this.sendBufferSize != null) { |
||||
|
driverConfigBuilder.withInt(DefaultDriverOption.SOCKET_SEND_BUFFER_SIZE, |
||||
|
this.sendBufferSize); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void initPoolingOptions(ProgrammaticDriverConfigLoaderBuilder driverConfigBuilder) { |
||||
|
driverConfigBuilder.withInt(DefaultDriverOption.CONNECTION_MAX_REQUESTS, |
||||
|
this.max_requests_local); |
||||
|
} |
||||
|
|
||||
|
private void initQueryOptions(ProgrammaticDriverConfigLoaderBuilder driverConfigBuilder) { |
||||
|
driverConfigBuilder.withInt(DefaultDriverOption.REQUEST_PAGE_SIZE, |
||||
|
this.defaultFetchSize); |
||||
|
} |
||||
|
|
||||
|
private List<String> getContactPoints(String url) { |
||||
|
List<String> result; |
||||
|
if (StringUtils.isBlank(url)) { |
||||
|
result = Collections.emptyList(); |
||||
|
} else { |
||||
|
result = new ArrayList<>(); |
||||
|
for (String hostPort : url.split(COMMA)) { |
||||
|
result.add(hostPort); |
||||
|
} |
||||
|
} |
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -1,73 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.cassandra; |
|
||||
|
|
||||
import com.datastax.driver.core.ConsistencyLevel; |
|
||||
import com.datastax.driver.core.QueryOptions; |
|
||||
import lombok.Data; |
|
||||
import org.springframework.beans.factory.annotation.Value; |
|
||||
import org.springframework.context.annotation.Configuration; |
|
||||
import org.springframework.stereotype.Component; |
|
||||
import org.thingsboard.server.dao.util.NoSqlAnyDao; |
|
||||
|
|
||||
import javax.annotation.PostConstruct; |
|
||||
|
|
||||
@Component |
|
||||
@Configuration |
|
||||
@Data |
|
||||
@NoSqlAnyDao |
|
||||
public class CassandraQueryOptions { |
|
||||
|
|
||||
@Value("${cassandra.query.default_fetch_size}") |
|
||||
private Integer defaultFetchSize; |
|
||||
@Value("${cassandra.query.read_consistency_level}") |
|
||||
private String readConsistencyLevel; |
|
||||
@Value("${cassandra.query.write_consistency_level}") |
|
||||
private String writeConsistencyLevel; |
|
||||
|
|
||||
private QueryOptions opts; |
|
||||
|
|
||||
private ConsistencyLevel defaultReadConsistencyLevel; |
|
||||
private ConsistencyLevel defaultWriteConsistencyLevel; |
|
||||
|
|
||||
@PostConstruct |
|
||||
public void initOpts(){ |
|
||||
opts = new QueryOptions(); |
|
||||
opts.setFetchSize(defaultFetchSize); |
|
||||
} |
|
||||
|
|
||||
protected ConsistencyLevel getDefaultReadConsistencyLevel() { |
|
||||
if (defaultReadConsistencyLevel == null) { |
|
||||
if (readConsistencyLevel != null) { |
|
||||
defaultReadConsistencyLevel = ConsistencyLevel.valueOf(readConsistencyLevel.toUpperCase()); |
|
||||
} else { |
|
||||
defaultReadConsistencyLevel = ConsistencyLevel.ONE; |
|
||||
} |
|
||||
} |
|
||||
return defaultReadConsistencyLevel; |
|
||||
} |
|
||||
|
|
||||
protected ConsistencyLevel getDefaultWriteConsistencyLevel() { |
|
||||
if (defaultWriteConsistencyLevel == null) { |
|
||||
if (writeConsistencyLevel != null) { |
|
||||
defaultWriteConsistencyLevel = ConsistencyLevel.valueOf(writeConsistencyLevel.toUpperCase()); |
|
||||
} else { |
|
||||
defaultWriteConsistencyLevel = ConsistencyLevel.ONE; |
|
||||
} |
|
||||
} |
|
||||
return defaultWriteConsistencyLevel; |
|
||||
} |
|
||||
} |
|
||||
@ -1,76 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.cassandra; |
|
||||
|
|
||||
import com.datastax.driver.core.SocketOptions; |
|
||||
import lombok.Data; |
|
||||
import org.springframework.beans.factory.annotation.Value; |
|
||||
import org.springframework.context.annotation.Configuration; |
|
||||
import org.springframework.stereotype.Component; |
|
||||
import org.thingsboard.server.dao.util.NoSqlAnyDao; |
|
||||
|
|
||||
import javax.annotation.PostConstruct; |
|
||||
|
|
||||
@Component |
|
||||
@Configuration |
|
||||
@Data |
|
||||
@NoSqlAnyDao |
|
||||
public class CassandraSocketOptions { |
|
||||
|
|
||||
@Value("${cassandra.socket.connect_timeout}") |
|
||||
private int connectTimeoutMillis; |
|
||||
@Value("${cassandra.socket.read_timeout}") |
|
||||
private int readTimeoutMillis; |
|
||||
@Value("${cassandra.socket.keep_alive}") |
|
||||
private Boolean keepAlive; |
|
||||
@Value("${cassandra.socket.reuse_address}") |
|
||||
private Boolean reuseAddress; |
|
||||
@Value("${cassandra.socket.so_linger}") |
|
||||
private Integer soLinger; |
|
||||
@Value("${cassandra.socket.tcp_no_delay}") |
|
||||
private Boolean tcpNoDelay; |
|
||||
@Value("${cassandra.socket.receive_buffer_size}") |
|
||||
private Integer receiveBufferSize; |
|
||||
@Value("${cassandra.socket.send_buffer_size}") |
|
||||
private Integer sendBufferSize; |
|
||||
|
|
||||
private SocketOptions opts; |
|
||||
|
|
||||
@PostConstruct |
|
||||
public void initOpts() { |
|
||||
opts = new SocketOptions(); |
|
||||
opts.setConnectTimeoutMillis(connectTimeoutMillis); |
|
||||
opts.setReadTimeoutMillis(readTimeoutMillis); |
|
||||
if (keepAlive != null) { |
|
||||
opts.setKeepAlive(keepAlive); |
|
||||
} |
|
||||
if (reuseAddress != null) { |
|
||||
opts.setReuseAddress(reuseAddress); |
|
||||
} |
|
||||
if (soLinger != null) { |
|
||||
opts.setSoLinger(soLinger); |
|
||||
} |
|
||||
if (tcpNoDelay != null) { |
|
||||
opts.setTcpNoDelay(tcpNoDelay); |
|
||||
} |
|
||||
if (receiveBufferSize != null) { |
|
||||
opts.setReceiveBufferSize(receiveBufferSize); |
|
||||
} |
|
||||
if (sendBufferSize != null) { |
|
||||
opts.setSendBufferSize(sendBufferSize); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,84 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 The Thingsboard Authors |
||||
|
* |
||||
|
* 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.dao.cassandra.guava; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.config.DriverConfigLoader; |
||||
|
import com.datastax.oss.driver.api.core.cql.PrepareRequest; |
||||
|
import com.datastax.oss.driver.api.core.cql.Statement; |
||||
|
import com.datastax.oss.driver.api.core.metadata.Node; |
||||
|
import com.datastax.oss.driver.api.core.metadata.NodeStateListener; |
||||
|
import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener; |
||||
|
import com.datastax.oss.driver.api.core.session.ProgrammaticArguments; |
||||
|
import com.datastax.oss.driver.api.core.tracker.RequestTracker; |
||||
|
import com.datastax.oss.driver.api.core.type.codec.TypeCodec; |
||||
|
import com.datastax.oss.driver.internal.core.context.DefaultDriverContext; |
||||
|
import com.datastax.oss.driver.internal.core.cql.CqlPrepareAsyncProcessor; |
||||
|
import com.datastax.oss.driver.internal.core.cql.CqlPrepareSyncProcessor; |
||||
|
import com.datastax.oss.driver.internal.core.cql.CqlRequestAsyncProcessor; |
||||
|
import com.datastax.oss.driver.internal.core.cql.CqlRequestSyncProcessor; |
||||
|
import com.datastax.oss.driver.internal.core.session.RequestProcessorRegistry; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.function.Predicate; |
||||
|
|
||||
|
/** |
||||
|
* A Custom {@link DefaultDriverContext} that overrides {@link #getRequestProcessorRegistry()} to |
||||
|
* return a {@link RequestProcessorRegistry} that includes processors for returning guava futures. |
||||
|
*/ |
||||
|
public class GuavaDriverContext extends DefaultDriverContext { |
||||
|
|
||||
|
public GuavaDriverContext( |
||||
|
DriverConfigLoader configLoader, |
||||
|
List<TypeCodec<?>> typeCodecs, |
||||
|
NodeStateListener nodeStateListener, |
||||
|
SchemaChangeListener schemaChangeListener, |
||||
|
RequestTracker requestTracker, |
||||
|
Map<String, String> localDatacenters, |
||||
|
Map<String, Predicate<Node>> nodeFilters, |
||||
|
ClassLoader classLoader) { |
||||
|
super( |
||||
|
configLoader, |
||||
|
ProgrammaticArguments.builder() |
||||
|
.addTypeCodecs(typeCodecs.toArray(new TypeCodec<?>[0])) |
||||
|
.withNodeStateListener(nodeStateListener) |
||||
|
.withSchemaChangeListener(schemaChangeListener) |
||||
|
.withRequestTracker(requestTracker) |
||||
|
.withLocalDatacenters(localDatacenters) |
||||
|
.withNodeFilters(nodeFilters) |
||||
|
.withClassLoader(classLoader) |
||||
|
.build()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public RequestProcessorRegistry buildRequestProcessorRegistry() { |
||||
|
// Register the typical request processors, except instead of the normal async processors,
|
||||
|
// use GuavaRequestAsyncProcessor to return ListenableFutures in async methods.
|
||||
|
|
||||
|
CqlRequestAsyncProcessor cqlRequestAsyncProcessor = new CqlRequestAsyncProcessor(); |
||||
|
CqlPrepareAsyncProcessor cqlPrepareAsyncProcessor = new CqlPrepareAsyncProcessor(); |
||||
|
CqlRequestSyncProcessor cqlRequestSyncProcessor = |
||||
|
new CqlRequestSyncProcessor(cqlRequestAsyncProcessor); |
||||
|
|
||||
|
return new RequestProcessorRegistry( |
||||
|
getSessionName(), |
||||
|
cqlRequestSyncProcessor, |
||||
|
new CqlPrepareSyncProcessor(cqlPrepareAsyncProcessor), |
||||
|
new GuavaRequestAsyncProcessor<>( |
||||
|
cqlRequestAsyncProcessor, Statement.class, GuavaSession.ASYNC), |
||||
|
new GuavaRequestAsyncProcessor<>( |
||||
|
cqlPrepareAsyncProcessor, PrepareRequest.class, GuavaSession.ASYNC_PREPARED)); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,79 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 The Thingsboard Authors |
||||
|
* |
||||
|
* 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.dao.cassandra.guava; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.session.Request; |
||||
|
import com.datastax.oss.driver.api.core.type.reflect.GenericType; |
||||
|
import com.datastax.oss.driver.internal.core.context.InternalDriverContext; |
||||
|
import com.datastax.oss.driver.internal.core.session.DefaultSession; |
||||
|
import com.datastax.oss.driver.internal.core.session.RequestProcessor; |
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import com.google.common.util.concurrent.SettableFuture; |
||||
|
import java.util.concurrent.CompletionStage; |
||||
|
|
||||
|
/** |
||||
|
* Wraps a {@link RequestProcessor} that returns {@link CompletionStage}s and converts them to a |
||||
|
* {@link ListenableFuture}s. |
||||
|
* |
||||
|
* @param <T> The type of request |
||||
|
* @param <U> The type of responses enclosed in the future response. |
||||
|
*/ |
||||
|
public class GuavaRequestAsyncProcessor<T extends Request, U> |
||||
|
implements RequestProcessor<T, ListenableFuture<U>> { |
||||
|
|
||||
|
private final RequestProcessor<T, CompletionStage<U>> subProcessor; |
||||
|
|
||||
|
private final GenericType resultType; |
||||
|
|
||||
|
private final Class<?> requestClass; |
||||
|
|
||||
|
GuavaRequestAsyncProcessor( |
||||
|
RequestProcessor<T, CompletionStage<U>> subProcessor, |
||||
|
Class<?> requestClass, |
||||
|
GenericType resultType) { |
||||
|
this.subProcessor = subProcessor; |
||||
|
this.requestClass = requestClass; |
||||
|
this.resultType = resultType; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean canProcess(Request request, GenericType resultType) { |
||||
|
return requestClass.isInstance(request) && resultType.equals(this.resultType); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public ListenableFuture<U> process( |
||||
|
T request, DefaultSession session, InternalDriverContext context, String sessionLogPrefix) { |
||||
|
SettableFuture<U> future = SettableFuture.create(); |
||||
|
subProcessor |
||||
|
.process(request, session, context, sessionLogPrefix) |
||||
|
.whenComplete( |
||||
|
(r, ex) -> { |
||||
|
if (ex != null) { |
||||
|
future.setException(ex); |
||||
|
} else { |
||||
|
future.set(r); |
||||
|
} |
||||
|
}); |
||||
|
return future; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public ListenableFuture<U> newFailure(RuntimeException error) { |
||||
|
return Futures.immediateFailedFuture(error); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,51 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 The Thingsboard Authors |
||||
|
* |
||||
|
* 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.dao.cassandra.guava; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.cql.AsyncResultSet; |
||||
|
import com.datastax.oss.driver.api.core.cql.PreparedStatement; |
||||
|
import com.datastax.oss.driver.api.core.cql.SimpleStatement; |
||||
|
import com.datastax.oss.driver.api.core.cql.Statement; |
||||
|
import com.datastax.oss.driver.api.core.cql.SyncCqlSession; |
||||
|
import com.datastax.oss.driver.api.core.session.Session; |
||||
|
import com.datastax.oss.driver.api.core.type.reflect.GenericType; |
||||
|
import com.datastax.oss.driver.internal.core.cql.DefaultPrepareRequest; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
|
||||
|
public interface GuavaSession extends Session, SyncCqlSession { |
||||
|
|
||||
|
GenericType<ListenableFuture<AsyncResultSet>> ASYNC = |
||||
|
new GenericType<ListenableFuture<AsyncResultSet>>() {}; |
||||
|
|
||||
|
GenericType<ListenableFuture<PreparedStatement>> ASYNC_PREPARED = |
||||
|
new GenericType<ListenableFuture<PreparedStatement>>() {}; |
||||
|
|
||||
|
default ListenableFuture<AsyncResultSet> executeAsync(Statement<?> statement) { |
||||
|
return this.execute(statement, ASYNC); |
||||
|
} |
||||
|
|
||||
|
default ListenableFuture<AsyncResultSet> executeAsync(String statement) { |
||||
|
return this.executeAsync(SimpleStatement.newInstance(statement)); |
||||
|
} |
||||
|
|
||||
|
default ListenableFuture<PreparedStatement> prepareAsync(SimpleStatement statement) { |
||||
|
return this.execute(new DefaultPrepareRequest(statement), ASYNC_PREPARED); |
||||
|
} |
||||
|
|
||||
|
default ListenableFuture<PreparedStatement> prepareAsync(String statement) { |
||||
|
return this.prepareAsync(SimpleStatement.newInstance(statement)); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,59 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 The Thingsboard Authors |
||||
|
* |
||||
|
* 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.dao.cassandra.guava; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.CqlSession; |
||||
|
import com.datastax.oss.driver.api.core.config.DriverConfigLoader; |
||||
|
import com.datastax.oss.driver.api.core.context.DriverContext; |
||||
|
import com.datastax.oss.driver.api.core.metadata.Node; |
||||
|
import com.datastax.oss.driver.api.core.metadata.NodeStateListener; |
||||
|
import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener; |
||||
|
import com.datastax.oss.driver.api.core.session.SessionBuilder; |
||||
|
import com.datastax.oss.driver.api.core.tracker.RequestTracker; |
||||
|
import com.datastax.oss.driver.api.core.type.codec.TypeCodec; |
||||
|
import edu.umd.cs.findbugs.annotations.NonNull; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.function.Predicate; |
||||
|
|
||||
|
public class GuavaSessionBuilder extends SessionBuilder<GuavaSessionBuilder, GuavaSession> { |
||||
|
|
||||
|
@Override |
||||
|
protected DriverContext buildContext( |
||||
|
DriverConfigLoader configLoader, |
||||
|
List<TypeCodec<?>> typeCodecs, |
||||
|
NodeStateListener nodeStateListener, |
||||
|
SchemaChangeListener schemaChangeListener, |
||||
|
RequestTracker requestTracker, |
||||
|
Map<String, String> localDatacenters, |
||||
|
Map<String, Predicate<Node>> nodeFilters, |
||||
|
ClassLoader classLoader) { |
||||
|
return new GuavaDriverContext( |
||||
|
configLoader, |
||||
|
typeCodecs, |
||||
|
nodeStateListener, |
||||
|
schemaChangeListener, |
||||
|
requestTracker, |
||||
|
localDatacenters, |
||||
|
nodeFilters, |
||||
|
classLoader); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected GuavaSession wrap(@NonNull CqlSession defaultSession) { |
||||
|
return new DefaultGuavaSession(defaultSession); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,131 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2020 The Thingsboard Authors |
||||
|
* |
||||
|
* 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.dao.nosql; |
||||
|
|
||||
|
import com.datastax.oss.driver.api.core.cql.AsyncResultSet; |
||||
|
import com.datastax.oss.driver.api.core.cql.ColumnDefinitions; |
||||
|
import com.datastax.oss.driver.api.core.cql.ExecutionInfo; |
||||
|
import com.datastax.oss.driver.api.core.cql.Row; |
||||
|
import com.datastax.oss.driver.api.core.cql.Statement; |
||||
|
import com.google.common.collect.Lists; |
||||
|
import com.google.common.util.concurrent.FutureCallback; |
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import com.google.common.util.concurrent.MoreExecutors; |
||||
|
import com.google.common.util.concurrent.SettableFuture; |
||||
|
import edu.umd.cs.findbugs.annotations.NonNull; |
||||
|
import org.checkerframework.checker.nullness.qual.Nullable; |
||||
|
|
||||
|
import java.nio.ByteBuffer; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.CompletionStage; |
||||
|
import java.util.concurrent.Executor; |
||||
|
import java.util.function.Function; |
||||
|
|
||||
|
public class TbResultSet implements AsyncResultSet { |
||||
|
|
||||
|
private final Statement originalStatement; |
||||
|
private final AsyncResultSet delegate; |
||||
|
private final Function<Statement, TbResultSetFuture> executeAsyncFunction; |
||||
|
|
||||
|
public TbResultSet(Statement originalStatement, AsyncResultSet delegate, |
||||
|
Function<Statement, TbResultSetFuture> executeAsyncFunction) { |
||||
|
this.originalStatement = originalStatement; |
||||
|
this.delegate = delegate; |
||||
|
this.executeAsyncFunction = executeAsyncFunction; |
||||
|
} |
||||
|
|
||||
|
@NonNull |
||||
|
@Override |
||||
|
public ColumnDefinitions getColumnDefinitions() { |
||||
|
return delegate.getColumnDefinitions(); |
||||
|
} |
||||
|
|
||||
|
@NonNull |
||||
|
@Override |
||||
|
public ExecutionInfo getExecutionInfo() { |
||||
|
return delegate.getExecutionInfo(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public int remaining() { |
||||
|
return delegate.remaining(); |
||||
|
} |
||||
|
|
||||
|
@NonNull |
||||
|
@Override |
||||
|
public Iterable<Row> currentPage() { |
||||
|
return delegate.currentPage(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean hasMorePages() { |
||||
|
return delegate.hasMorePages(); |
||||
|
} |
||||
|
|
||||
|
@NonNull |
||||
|
@Override |
||||
|
public CompletionStage<AsyncResultSet> fetchNextPage() throws IllegalStateException { |
||||
|
return delegate.fetchNextPage(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean wasApplied() { |
||||
|
return delegate.wasApplied(); |
||||
|
} |
||||
|
|
||||
|
public ListenableFuture<List<Row>> allRows(Executor executor) { |
||||
|
List<Row> allRows = new ArrayList<>(); |
||||
|
SettableFuture<List<Row>> resultFuture = SettableFuture.create(); |
||||
|
this.processRows(originalStatement, delegate, allRows, resultFuture, executor); |
||||
|
return resultFuture; |
||||
|
} |
||||
|
|
||||
|
private void processRows(Statement statement, |
||||
|
AsyncResultSet resultSet, |
||||
|
List<Row> allRows, |
||||
|
SettableFuture<List<Row>> resultFuture, |
||||
|
Executor executor) { |
||||
|
allRows.addAll(loadRows(resultSet)); |
||||
|
if (resultSet.hasMorePages()) { |
||||
|
ByteBuffer nextPagingState = resultSet.getExecutionInfo().getPagingState(); |
||||
|
Statement<?> nextStatement = statement.setPagingState(nextPagingState); |
||||
|
TbResultSetFuture resultSetFuture = executeAsyncFunction.apply(nextStatement); |
||||
|
Futures.addCallback(resultSetFuture, |
||||
|
new FutureCallback<TbResultSet>() { |
||||
|
@Override |
||||
|
public void onSuccess(@Nullable TbResultSet result) { |
||||
|
processRows(nextStatement, result, |
||||
|
allRows, resultFuture, executor); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
resultFuture.setException(t); |
||||
|
} |
||||
|
}, executor != null ? executor : MoreExecutors.directExecutor() |
||||
|
); |
||||
|
} else { |
||||
|
resultFuture.set(allRows); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
List<Row> loadRows(AsyncResultSet resultSet) { |
||||
|
return Lists.newArrayList(resultSet.currentPage()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -1,121 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.model; |
|
||||
|
|
||||
import com.datastax.driver.mapping.annotations.Column; |
|
||||
import com.datastax.driver.mapping.annotations.PartitionKey; |
|
||||
import com.datastax.driver.mapping.annotations.Table; |
|
||||
import org.thingsboard.server.common.data.EntitySubtype; |
|
||||
import org.thingsboard.server.common.data.EntityType; |
|
||||
import org.thingsboard.server.common.data.id.TenantId; |
|
||||
import org.thingsboard.server.dao.model.type.EntityTypeCodec; |
|
||||
|
|
||||
import java.util.UUID; |
|
||||
|
|
||||
import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_SUBTYPE_COLUMN_FAMILY_NAME; |
|
||||
import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_SUBTYPE_ENTITY_TYPE_PROPERTY; |
|
||||
import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_SUBTYPE_TENANT_ID_PROPERTY; |
|
||||
import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_SUBTYPE_TYPE_PROPERTY; |
|
||||
|
|
||||
@Table(name = ENTITY_SUBTYPE_COLUMN_FAMILY_NAME) |
|
||||
public class EntitySubtypeEntity { |
|
||||
|
|
||||
@PartitionKey(value = 0) |
|
||||
@Column(name = ENTITY_SUBTYPE_TENANT_ID_PROPERTY) |
|
||||
private UUID tenantId; |
|
||||
|
|
||||
@PartitionKey(value = 1) |
|
||||
@Column(name = ENTITY_SUBTYPE_ENTITY_TYPE_PROPERTY, codec = EntityTypeCodec.class) |
|
||||
private EntityType entityType; |
|
||||
|
|
||||
@PartitionKey(value = 2) |
|
||||
@Column(name = ENTITY_SUBTYPE_TYPE_PROPERTY) |
|
||||
private String type; |
|
||||
|
|
||||
public EntitySubtypeEntity() { |
|
||||
super(); |
|
||||
} |
|
||||
|
|
||||
public EntitySubtypeEntity(EntitySubtype entitySubtype) { |
|
||||
this.tenantId = entitySubtype.getTenantId().getId(); |
|
||||
this.entityType = entitySubtype.getEntityType(); |
|
||||
this.type = entitySubtype.getType(); |
|
||||
} |
|
||||
|
|
||||
public UUID getTenantId() { |
|
||||
return tenantId; |
|
||||
} |
|
||||
|
|
||||
public void setTenantId(UUID tenantId) { |
|
||||
this.tenantId = tenantId; |
|
||||
} |
|
||||
|
|
||||
public EntityType getEntityType() { |
|
||||
return entityType; |
|
||||
} |
|
||||
|
|
||||
public void setEntityType(EntityType entityType) { |
|
||||
this.entityType = entityType; |
|
||||
} |
|
||||
|
|
||||
public String getType() { |
|
||||
return type; |
|
||||
} |
|
||||
|
|
||||
public void setType(String type) { |
|
||||
this.type = type; |
|
||||
} |
|
||||
|
|
||||
|
|
||||
@Override |
|
||||
public boolean equals(Object o) { |
|
||||
if (this == o) return true; |
|
||||
if (o == null || getClass() != o.getClass()) return false; |
|
||||
|
|
||||
EntitySubtypeEntity that = (EntitySubtypeEntity) o; |
|
||||
|
|
||||
if (tenantId != null ? !tenantId.equals(that.tenantId) : that.tenantId != null) return false; |
|
||||
if (entityType != that.entityType) return false; |
|
||||
return type != null ? type.equals(that.type) : that.type == null; |
|
||||
|
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public int hashCode() { |
|
||||
int result = tenantId != null ? tenantId.hashCode() : 0; |
|
||||
result = 31 * result + (entityType != null ? entityType.hashCode() : 0); |
|
||||
result = 31 * result + (type != null ? type.hashCode() : 0); |
|
||||
return result; |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public String toString() { |
|
||||
final StringBuilder sb = new StringBuilder("EntitySubtypeEntity{"); |
|
||||
sb.append("tenantId=").append(tenantId); |
|
||||
sb.append(", entityType=").append(entityType); |
|
||||
sb.append(", type='").append(type).append('\''); |
|
||||
sb.append('}'); |
|
||||
return sb.toString(); |
|
||||
} |
|
||||
|
|
||||
public EntitySubtype toEntitySubtype() { |
|
||||
EntitySubtype entitySubtype = new EntitySubtype(); |
|
||||
entitySubtype.setTenantId(new TenantId(tenantId)); |
|
||||
entitySubtype.setEntityType(entityType); |
|
||||
entitySubtype.setType(type); |
|
||||
return entitySubtype; |
|
||||
} |
|
||||
} |
|
||||
@ -1,27 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.model.type; |
|
||||
|
|
||||
import com.datastax.driver.extras.codecs.enums.EnumNameCodec; |
|
||||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleState; |
|
||||
|
|
||||
public class ComponentLifecycleStateCodec extends EnumNameCodec<ComponentLifecycleState> { |
|
||||
|
|
||||
public ComponentLifecycleStateCodec() { |
|
||||
super(ComponentLifecycleState.class); |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -1,38 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.model.wrapper; |
|
||||
|
|
||||
|
|
||||
import com.datastax.driver.core.ResultSet; |
|
||||
|
|
||||
public class EntityResultSet<T> { |
|
||||
|
|
||||
private ResultSet resultSet; |
|
||||
private T entity; |
|
||||
|
|
||||
public EntityResultSet(ResultSet resultSet, T entity) { |
|
||||
this.resultSet = resultSet; |
|
||||
this.entity = entity; |
|
||||
} |
|
||||
|
|
||||
public T getEntity() { |
|
||||
return entity; |
|
||||
} |
|
||||
|
|
||||
public boolean wasApplied() { |
|
||||
return resultSet.wasApplied(); |
|
||||
} |
|
||||
} |
|
||||
@ -1,156 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.nosql; |
|
||||
|
|
||||
import com.datastax.driver.core.ResultSet; |
|
||||
import com.datastax.driver.core.ResultSetFuture; |
|
||||
import com.datastax.driver.core.Session; |
|
||||
import com.datastax.driver.core.Statement; |
|
||||
import com.google.common.util.concurrent.FutureCallback; |
|
||||
import com.google.common.util.concurrent.Futures; |
|
||||
import com.google.common.util.concurrent.ListenableFuture; |
|
||||
import com.google.common.util.concurrent.MoreExecutors; |
|
||||
import com.google.common.util.concurrent.Uninterruptibles; |
|
||||
import org.thingsboard.server.dao.exception.BufferLimitException; |
|
||||
import org.thingsboard.server.dao.util.AsyncRateLimiter; |
|
||||
|
|
||||
import javax.annotation.Nullable; |
|
||||
import java.util.concurrent.CancellationException; |
|
||||
import java.util.concurrent.ExecutionException; |
|
||||
import java.util.concurrent.Executor; |
|
||||
import java.util.concurrent.TimeUnit; |
|
||||
import java.util.concurrent.TimeoutException; |
|
||||
|
|
||||
public class RateLimitedResultSetFuture implements ResultSetFuture { |
|
||||
|
|
||||
private final ListenableFuture<ResultSetFuture> originalFuture; |
|
||||
private final ListenableFuture<Void> rateLimitFuture; |
|
||||
|
|
||||
public RateLimitedResultSetFuture(Session session, AsyncRateLimiter rateLimiter, Statement statement) { |
|
||||
this.rateLimitFuture = Futures.catchingAsync(rateLimiter.acquireAsync(), Throwable.class, t -> { |
|
||||
if (!(t instanceof BufferLimitException)) { |
|
||||
rateLimiter.release(); |
|
||||
} |
|
||||
return Futures.immediateFailedFuture(t); |
|
||||
}, MoreExecutors.directExecutor()); |
|
||||
this.originalFuture = Futures.transform(rateLimitFuture, |
|
||||
i -> executeAsyncWithRelease(rateLimiter, session, statement), MoreExecutors.directExecutor()); |
|
||||
|
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public ResultSet getUninterruptibly() { |
|
||||
return safeGet().getUninterruptibly(); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public ResultSet getUninterruptibly(long timeout, TimeUnit unit) throws TimeoutException { |
|
||||
long rateLimitStart = System.nanoTime(); |
|
||||
ResultSetFuture resultSetFuture = null; |
|
||||
try { |
|
||||
resultSetFuture = originalFuture.get(timeout, unit); |
|
||||
} catch (InterruptedException | ExecutionException e) { |
|
||||
throw new IllegalStateException(e); |
|
||||
} |
|
||||
long rateLimitDurationNano = System.nanoTime() - rateLimitStart; |
|
||||
long innerTimeoutNano = unit.toNanos(timeout) - rateLimitDurationNano; |
|
||||
if (innerTimeoutNano > 0) { |
|
||||
return resultSetFuture.getUninterruptibly(innerTimeoutNano, TimeUnit.NANOSECONDS); |
|
||||
} |
|
||||
throw new TimeoutException("Timeout waiting for task."); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public boolean cancel(boolean mayInterruptIfRunning) { |
|
||||
if (originalFuture.isDone()) { |
|
||||
return safeGet().cancel(mayInterruptIfRunning); |
|
||||
} else { |
|
||||
return originalFuture.cancel(mayInterruptIfRunning); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public boolean isCancelled() { |
|
||||
if (originalFuture.isDone()) { |
|
||||
return safeGet().isCancelled(); |
|
||||
} |
|
||||
|
|
||||
return originalFuture.isCancelled(); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public boolean isDone() { |
|
||||
return originalFuture.isDone() && safeGet().isDone(); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public ResultSet get() throws InterruptedException, ExecutionException { |
|
||||
return safeGet().get(); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public ResultSet get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { |
|
||||
long rateLimitStart = System.nanoTime(); |
|
||||
ResultSetFuture resultSetFuture = originalFuture.get(timeout, unit); |
|
||||
long rateLimitDurationNano = System.nanoTime() - rateLimitStart; |
|
||||
long innerTimeoutNano = unit.toNanos(timeout) - rateLimitDurationNano; |
|
||||
if (innerTimeoutNano > 0) { |
|
||||
return resultSetFuture.get(innerTimeoutNano, TimeUnit.NANOSECONDS); |
|
||||
} |
|
||||
throw new TimeoutException("Timeout waiting for task."); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void addListener(Runnable listener, Executor executor) { |
|
||||
originalFuture.addListener(() -> { |
|
||||
try { |
|
||||
ResultSetFuture resultSetFuture = Uninterruptibles.getUninterruptibly(originalFuture); |
|
||||
resultSetFuture.addListener(listener, executor); |
|
||||
} catch (CancellationException | ExecutionException e) { |
|
||||
Futures.immediateFailedFuture(e).addListener(listener, executor); |
|
||||
} |
|
||||
}, executor); |
|
||||
} |
|
||||
|
|
||||
private ResultSetFuture safeGet() { |
|
||||
try { |
|
||||
return originalFuture.get(); |
|
||||
} catch (InterruptedException | ExecutionException e) { |
|
||||
throw new IllegalStateException(e); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
private ResultSetFuture executeAsyncWithRelease(AsyncRateLimiter rateLimiter, Session session, Statement statement) { |
|
||||
try { |
|
||||
ResultSetFuture resultSetFuture = session.executeAsync(statement); |
|
||||
Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() { |
|
||||
@Override |
|
||||
public void onSuccess(@Nullable ResultSet result) { |
|
||||
rateLimiter.release(); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void onFailure(Throwable t) { |
|
||||
rateLimiter.release(); |
|
||||
} |
|
||||
}, MoreExecutors.directExecutor()); |
|
||||
return resultSetFuture; |
|
||||
} catch (RuntimeException re) { |
|
||||
rateLimiter.release(); |
|
||||
throw re; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -1,195 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2020 The Thingsboard Authors |
|
||||
* |
|
||||
* 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. |
|
||||
*/ |
|
||||
package org.thingsboard.server.dao.nosql; |
|
||||
|
|
||||
import com.datastax.driver.core.ProtocolVersion; |
|
||||
import com.datastax.driver.core.ResultSet; |
|
||||
import com.datastax.driver.core.ResultSetFuture; |
|
||||
import com.datastax.driver.core.Row; |
|
||||
import com.datastax.driver.core.Session; |
|
||||
import com.datastax.driver.core.Statement; |
|
||||
import com.datastax.driver.core.exceptions.UnsupportedFeatureException; |
|
||||
import com.google.common.util.concurrent.Futures; |
|
||||
import com.google.common.util.concurrent.ListenableFuture; |
|
||||
import com.google.common.util.concurrent.MoreExecutors; |
|
||||
import org.junit.Test; |
|
||||
import org.junit.runner.RunWith; |
|
||||
import org.mockito.Mock; |
|
||||
import org.mockito.Mockito; |
|
||||
import org.mockito.runners.MockitoJUnitRunner; |
|
||||
import org.mockito.stubbing.Answer; |
|
||||
import org.thingsboard.server.dao.exception.BufferLimitException; |
|
||||
import org.thingsboard.server.dao.util.AsyncRateLimiter; |
|
||||
|
|
||||
import java.util.concurrent.CountDownLatch; |
|
||||
import java.util.concurrent.ExecutionException; |
|
||||
import java.util.concurrent.Executors; |
|
||||
import java.util.concurrent.TimeoutException; |
|
||||
|
|
||||
import static org.junit.Assert.assertSame; |
|
||||
import static org.junit.Assert.assertTrue; |
|
||||
import static org.junit.Assert.fail; |
|
||||
import static org.mockito.Mockito.times; |
|
||||
import static org.mockito.Mockito.verify; |
|
||||
import static org.mockito.Mockito.verifyNoMoreInteractions; |
|
||||
import static org.mockito.Mockito.when; |
|
||||
|
|
||||
@RunWith(MockitoJUnitRunner.class) |
|
||||
public class RateLimitedResultSetFutureTest { |
|
||||
|
|
||||
private RateLimitedResultSetFuture resultSetFuture; |
|
||||
|
|
||||
@Mock |
|
||||
private AsyncRateLimiter rateLimiter; |
|
||||
@Mock |
|
||||
private Session session; |
|
||||
@Mock |
|
||||
private Statement statement; |
|
||||
@Mock |
|
||||
private ResultSetFuture realFuture; |
|
||||
@Mock |
|
||||
private ResultSet rows; |
|
||||
@Mock |
|
||||
private Row row; |
|
||||
|
|
||||
@Test |
|
||||
public void doNotReleasePermissionIfRateLimitFutureFailed() throws InterruptedException { |
|
||||
when(rateLimiter.acquireAsync()).thenReturn(Futures.immediateFailedFuture(new BufferLimitException())); |
|
||||
resultSetFuture = new RateLimitedResultSetFuture(session, rateLimiter, statement); |
|
||||
Thread.sleep(1000L); |
|
||||
verify(rateLimiter).acquireAsync(); |
|
||||
try { |
|
||||
assertTrue(resultSetFuture.isDone()); |
|
||||
fail(); |
|
||||
} catch (Exception e) { |
|
||||
assertTrue(e instanceof IllegalStateException); |
|
||||
Throwable actualCause = e.getCause(); |
|
||||
assertTrue(actualCause instanceof ExecutionException); |
|
||||
} |
|
||||
verifyNoMoreInteractions(session, rateLimiter, statement); |
|
||||
|
|
||||
} |
|
||||
|
|
||||
@Test |
|
||||
public void getUninterruptiblyDelegateToCassandra() throws InterruptedException, ExecutionException { |
|
||||
when(rateLimiter.acquireAsync()).thenReturn(Futures.immediateFuture(null)); |
|
||||
when(session.executeAsync(statement)).thenReturn(realFuture); |
|
||||
Mockito.doAnswer((Answer<Void>) invocation -> { |
|
||||
Object[] args = invocation.getArguments(); |
|
||||
Runnable task = (Runnable) args[0]; |
|
||||
task.run(); |
|
||||
return null; |
|
||||
}).when(realFuture).addListener(Mockito.any(), Mockito.any()); |
|
||||
|
|
||||
when(realFuture.getUninterruptibly()).thenReturn(rows); |
|
||||
|
|
||||
resultSetFuture = new RateLimitedResultSetFuture(session, rateLimiter, statement); |
|
||||
ResultSet actual = resultSetFuture.getUninterruptibly(); |
|
||||
assertSame(rows, actual); |
|
||||
verify(rateLimiter, times(1)).acquireAsync(); |
|
||||
verify(rateLimiter, times(1)).release(); |
|
||||
} |
|
||||
|
|
||||
@Test |
|
||||
public void addListenerAllowsFutureTransformation() throws InterruptedException, ExecutionException { |
|
||||
when(rateLimiter.acquireAsync()).thenReturn(Futures.immediateFuture(null)); |
|
||||
when(session.executeAsync(statement)).thenReturn(realFuture); |
|
||||
Mockito.doAnswer((Answer<Void>) invocation -> { |
|
||||
Object[] args = invocation.getArguments(); |
|
||||
Runnable task = (Runnable) args[0]; |
|
||||
task.run(); |
|
||||
return null; |
|
||||
}).when(realFuture).addListener(Mockito.any(), Mockito.any()); |
|
||||
|
|
||||
when(realFuture.get()).thenReturn(rows); |
|
||||
when(rows.one()).thenReturn(row); |
|
||||
|
|
||||
resultSetFuture = new RateLimitedResultSetFuture(session, rateLimiter, statement); |
|
||||
|
|
||||
ListenableFuture<Row> transform = Futures.transform(resultSetFuture, ResultSet::one, MoreExecutors.directExecutor()); |
|
||||
Row actualRow = transform.get(); |
|
||||
|
|
||||
assertSame(row, actualRow); |
|
||||
verify(rateLimiter, times(1)).acquireAsync(); |
|
||||
verify(rateLimiter, times(1)).release(); |
|
||||
} |
|
||||
|
|
||||
@Test |
|
||||
public void immidiateCassandraExceptionReturnsPermit() throws InterruptedException, ExecutionException { |
|
||||
when(rateLimiter.acquireAsync()).thenReturn(Futures.immediateFuture(null)); |
|
||||
when(session.executeAsync(statement)).thenThrow(new UnsupportedFeatureException(ProtocolVersion.V3, "hjg")); |
|
||||
resultSetFuture = new RateLimitedResultSetFuture(session, rateLimiter, statement); |
|
||||
ListenableFuture<Row> transform = Futures.transform(resultSetFuture, ResultSet::one, MoreExecutors.directExecutor()); |
|
||||
try { |
|
||||
transform.get(); |
|
||||
fail(); |
|
||||
} catch (Exception e) { |
|
||||
assertTrue(e instanceof ExecutionException); |
|
||||
} |
|
||||
verify(rateLimiter, times(1)).acquireAsync(); |
|
||||
verify(rateLimiter, times(1)).release(); |
|
||||
} |
|
||||
|
|
||||
@Test |
|
||||
public void queryTimeoutReturnsPermit() throws InterruptedException, ExecutionException { |
|
||||
when(rateLimiter.acquireAsync()).thenReturn(Futures.immediateFuture(null)); |
|
||||
when(session.executeAsync(statement)).thenReturn(realFuture); |
|
||||
Mockito.doAnswer((Answer<Void>) invocation -> { |
|
||||
Object[] args = invocation.getArguments(); |
|
||||
Runnable task = (Runnable) args[0]; |
|
||||
task.run(); |
|
||||
return null; |
|
||||
}).when(realFuture).addListener(Mockito.any(), Mockito.any()); |
|
||||
|
|
||||
when(realFuture.get()).thenThrow(new ExecutionException("Fail", new TimeoutException("timeout"))); |
|
||||
resultSetFuture = new RateLimitedResultSetFuture(session, rateLimiter, statement); |
|
||||
ListenableFuture<Row> transform = Futures.transform(resultSetFuture, ResultSet::one, MoreExecutors.directExecutor()); |
|
||||
try { |
|
||||
transform.get(); |
|
||||
fail(); |
|
||||
} catch (Exception e) { |
|
||||
assertTrue(e instanceof ExecutionException); |
|
||||
} |
|
||||
verify(rateLimiter, times(1)).acquireAsync(); |
|
||||
verify(rateLimiter, times(1)).release(); |
|
||||
} |
|
||||
|
|
||||
@Test |
|
||||
public void expiredQueryReturnPermit() throws InterruptedException, ExecutionException { |
|
||||
CountDownLatch latch = new CountDownLatch(1); |
|
||||
ListenableFuture<Void> future = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1)).submit(() -> { |
|
||||
latch.await(); |
|
||||
return null; |
|
||||
}); |
|
||||
when(rateLimiter.acquireAsync()).thenReturn(future); |
|
||||
resultSetFuture = new RateLimitedResultSetFuture(session, rateLimiter, statement); |
|
||||
|
|
||||
ListenableFuture<Row> transform = Futures.transform(resultSetFuture, ResultSet::one, MoreExecutors.directExecutor()); |
|
||||
// TimeUnit.MILLISECONDS.sleep(200);
|
|
||||
future.cancel(false); |
|
||||
latch.countDown(); |
|
||||
|
|
||||
try { |
|
||||
transform.get(); |
|
||||
fail(); |
|
||||
} catch (Exception e) { |
|
||||
assertTrue(e instanceof ExecutionException); |
|
||||
} |
|
||||
verify(rateLimiter, times(1)).acquireAsync(); |
|
||||
verify(rateLimiter, times(1)).release(); |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue