Igor Kulikov
6 years ago
110 changed files with 1337 additions and 1502 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); |
|||
} |
|||
} |
|||
@ -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; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,65 @@ |
|||
/** |
|||
* 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.Row; |
|||
import com.google.common.collect.Lists; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.SettableFuture; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.concurrent.CompletionStage; |
|||
import java.util.concurrent.Executor; |
|||
import java.util.stream.Collectors; |
|||
|
|||
public class ResultSetUtils { |
|||
|
|||
public static ListenableFuture<List<Row>> allRows(AsyncResultSet resultSet, Executor executor) { |
|||
List<ListenableFuture<AsyncResultSet>> futures = new ArrayList<>(); |
|||
futures.add(Futures.immediateFuture(resultSet)); |
|||
while (resultSet.hasMorePages()) { |
|||
futures.add(toListenable(resultSet.fetchNextPage())); |
|||
} |
|||
return Futures.transform( Futures.allAsList(futures), |
|||
resultSets -> resultSets.stream() |
|||
.map(rs -> loadRows(rs)) |
|||
.flatMap(rows -> rows.stream()) |
|||
.collect(Collectors.toList()), |
|||
executor |
|||
); |
|||
} |
|||
|
|||
private static <T> ListenableFuture<T> toListenable(CompletionStage<T> completable) { |
|||
SettableFuture<T> future = SettableFuture.create(); |
|||
completable.whenComplete( |
|||
(r, ex) -> { |
|||
if (ex != null) { |
|||
future.setException(ex); |
|||
} else { |
|||
future.set(r); |
|||
} |
|||
} |
|||
); |
|||
return future; |
|||
} |
|||
|
|||
private static List<Row> loadRows(AsyncResultSet resultSet) { |
|||
return Lists.newArrayList(resultSet.currentPage()); |
|||
} |
|||
} |
|||
@ -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