Andrew Shvayka
2 years ago
committed by
GitHub
GitHub
34 changed files with 658 additions and 392 deletions
@ -0,0 +1,177 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 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.service.cf; |
||||
|
|
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.thingsboard.server.actors.ActorSystemContext; |
||||
|
import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg; |
||||
|
import org.thingsboard.server.common.data.StringUtils; |
||||
|
import org.thingsboard.server.common.data.cf.CalculatedFieldType; |
||||
|
import org.thingsboard.server.common.data.id.CalculatedFieldId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.EntityIdFactory; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.kv.BasicKvEntry; |
||||
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
||||
|
import org.thingsboard.server.common.util.KvProtoUtil; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateMsgProto; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.TsDoubleValProto; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos.TsRollingArgumentProto; |
||||
|
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; |
||||
|
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
||||
|
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; |
||||
|
import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState; |
||||
|
import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState; |
||||
|
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; |
||||
|
import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry; |
||||
|
|
||||
|
import java.util.Optional; |
||||
|
import java.util.TreeMap; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
public abstract class AbstractCalculatedFieldStateService implements CalculatedFieldStateService { |
||||
|
|
||||
|
@Autowired |
||||
|
private ActorSystemContext actorSystemContext; |
||||
|
|
||||
|
@Override |
||||
|
public final void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { |
||||
|
CalculatedFieldStateMsgProto stateMsg = toProto(stateId, state); |
||||
|
long maxStateSizeInKBytes = ctx.getMaxStateSizeInKBytes(); |
||||
|
if (maxStateSizeInKBytes <= 0 || stateMsg.getSerializedSize() <= maxStateSizeInKBytes) { |
||||
|
doPersist(stateId, stateMsg, callback); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
protected abstract void doPersist(CalculatedFieldEntityCtxId stateId, CalculatedFieldStateMsgProto stateMsgProto, TbCallback callback); |
||||
|
|
||||
|
@Override |
||||
|
public final void removeState(CalculatedFieldEntityCtxId stateId, TbCallback callback) { |
||||
|
doRemove(stateId, callback); |
||||
|
} |
||||
|
|
||||
|
protected abstract void doRemove(CalculatedFieldEntityCtxId stateId, TbCallback callback); |
||||
|
|
||||
|
protected void processRestoredState(CalculatedFieldStateMsgProto stateMsg) { |
||||
|
CalculatedFieldEntityCtxId stateId = fromProto(stateMsg.getId()); |
||||
|
CalculatedFieldState state = stateMsg.hasState() ? fromProto(stateMsg.getState()) : null; |
||||
|
actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(stateId, state)); |
||||
|
} |
||||
|
|
||||
|
protected CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { |
||||
|
return CalculatedFieldEntityCtxIdProto.newBuilder() |
||||
|
.setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits()) |
||||
|
.setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) |
||||
|
.setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) |
||||
|
.setEntityType(ctxId.entityId().getEntityType().name()) |
||||
|
.setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) |
||||
|
.setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
protected CalculatedFieldEntityCtxId fromProto(CalculatedFieldEntityCtxIdProto ctxIdProto) { |
||||
|
TenantId tenantId = TenantId.fromUUID(new UUID(ctxIdProto.getTenantIdMSB(), ctxIdProto.getTenantIdLSB())); |
||||
|
EntityId entityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); |
||||
|
CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); |
||||
|
return new CalculatedFieldEntityCtxId(tenantId, calculatedFieldId, entityId); |
||||
|
} |
||||
|
|
||||
|
protected CalculatedFieldStateMsgProto toProto(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state) { |
||||
|
var stateProto = CalculatedFieldStateProto.newBuilder() |
||||
|
.setType(state.getType().name()); |
||||
|
state.getArguments().forEach((argName, argEntry) -> { |
||||
|
if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { |
||||
|
stateProto.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry)); |
||||
|
} else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) { |
||||
|
stateProto.addRollingValueArguments(toRollingArgumentProto(argName, rollingArgumentEntry)); |
||||
|
} |
||||
|
}); |
||||
|
return CalculatedFieldStateMsgProto.newBuilder() |
||||
|
.setId(toProto(stateId)) |
||||
|
.setState(stateProto) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
private TransportProtos.SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) { |
||||
|
TransportProtos.SingleValueArgumentProto.Builder builder = TransportProtos.SingleValueArgumentProto.newBuilder() |
||||
|
.setArgName(argName); |
||||
|
|
||||
|
if (entry != SingleValueArgumentEntry.EMPTY) { |
||||
|
builder.setValue(KvProtoUtil.toTsValueProto(entry.getTs(), entry.getKvEntryValue())); |
||||
|
} |
||||
|
|
||||
|
Optional.ofNullable(entry.getVersion()).ifPresent(builder::setVersion); |
||||
|
|
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
private TsRollingArgumentProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) { |
||||
|
TsRollingArgumentProto.Builder builder = TsRollingArgumentProto.newBuilder().setKey(argName); |
||||
|
|
||||
|
if (entry != TsRollingArgumentEntry.EMPTY) { |
||||
|
entry.getTsRecords().forEach((ts, value) -> builder.addTsValue(TsDoubleValProto.newBuilder().setTs(ts).setValue(value).build())); |
||||
|
} |
||||
|
|
||||
|
return builder.build(); |
||||
|
} |
||||
|
|
||||
|
private CalculatedFieldState fromProto(CalculatedFieldStateProto proto) { |
||||
|
if (StringUtils.isEmpty(proto.getType())) { |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
CalculatedFieldType type = CalculatedFieldType.valueOf(proto.getType()); |
||||
|
|
||||
|
CalculatedFieldState state = switch (type) { |
||||
|
case SIMPLE -> new SimpleCalculatedFieldState(); |
||||
|
case SCRIPT -> new ScriptCalculatedFieldState(); |
||||
|
}; |
||||
|
|
||||
|
proto.getSingleValueArgumentsList().forEach(argProto -> |
||||
|
state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto))); |
||||
|
|
||||
|
if (CalculatedFieldType.SCRIPT.equals(type)) { |
||||
|
proto.getRollingValueArgumentsList().forEach(argProto -> |
||||
|
state.getArguments().put(argProto.getKey(), fromRollingArgumentProto(argProto))); |
||||
|
} |
||||
|
|
||||
|
return state; |
||||
|
} |
||||
|
|
||||
|
private SingleValueArgumentEntry fromSingleValueArgumentProto(TransportProtos.SingleValueArgumentProto proto) { |
||||
|
if (!proto.hasValue()) { |
||||
|
return (SingleValueArgumentEntry) SingleValueArgumentEntry.EMPTY; |
||||
|
} |
||||
|
TransportProtos.TsValueProto tsValueProto = proto.getValue(); |
||||
|
long ts = tsValueProto.getTs(); |
||||
|
BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getArgName(), tsValueProto); |
||||
|
return new SingleValueArgumentEntry(ts, kvEntry, proto.getVersion()); |
||||
|
} |
||||
|
|
||||
|
private TsRollingArgumentEntry fromRollingArgumentProto(TsRollingArgumentProto proto) { |
||||
|
if (proto.getTsValueCount() <= 0) { |
||||
|
return (TsRollingArgumentEntry) TsRollingArgumentEntry.EMPTY; |
||||
|
} |
||||
|
TreeMap<Long, Double> tsRecords = new TreeMap<>(); |
||||
|
proto.getTsValueList().forEach(tsValueProto -> tsRecords.put(tsValueProto.getTs(), tsValueProto.getValue())); |
||||
|
return new TsRollingArgumentEntry(tsRecords); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -1,97 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2024 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.service.cf; |
|
||||
|
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.rocksdb.RocksDB; |
|
||||
import org.rocksdb.RocksDBException; |
|
||||
import org.rocksdb.RocksIterator; |
|
||||
import org.rocksdb.WriteOptions; |
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|
||||
import org.springframework.stereotype.Service; |
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; |
|
||||
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; |
|
||||
import org.thingsboard.server.utils.RocksDBConfig; |
|
||||
|
|
||||
import java.nio.charset.StandardCharsets; |
|
||||
import java.util.HashMap; |
|
||||
import java.util.Map; |
|
||||
|
|
||||
@Service |
|
||||
@Slf4j |
|
||||
@ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true) |
|
||||
public class RocksDBService { |
|
||||
|
|
||||
private final RocksDB db; |
|
||||
private final WriteOptions writeOptions; |
|
||||
|
|
||||
public RocksDBService(RocksDBConfig config) throws RocksDBException { |
|
||||
this.db = config.getDb(); |
|
||||
this.writeOptions = new WriteOptions().setSync(true); |
|
||||
} |
|
||||
|
|
||||
public void put(String key, String value) { |
|
||||
try { |
|
||||
db.put(writeOptions, key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8)); |
|
||||
} catch (RocksDBException e) { |
|
||||
log.error("Failed to store data to RocksDB", e); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
public void put(CalculatedFieldEntityCtxIdProto key, CalculatedFieldStateProto value) { |
|
||||
try { |
|
||||
db.put(writeOptions, key.toByteArray(), value.toByteArray()); |
|
||||
} catch (RocksDBException e) { |
|
||||
log.error("Failed to store data to RocksDB", e); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
public void delete(CalculatedFieldEntityCtxIdProto key) { |
|
||||
try { |
|
||||
db.delete(writeOptions, key.toByteArray()); |
|
||||
} catch (RocksDBException e) { |
|
||||
log.error("Failed to delete data from RocksDB", e); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
public String get(String key) { |
|
||||
try { |
|
||||
byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8)); |
|
||||
return value != null ? new String(value, StandardCharsets.UTF_8) : null; |
|
||||
} catch (RocksDBException e) { |
|
||||
log.error("Failed to retrieve data from RocksDB", e); |
|
||||
return null; |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
public Map<CalculatedFieldEntityCtxIdProto, CalculatedFieldStateProto> getAll() { |
|
||||
Map<CalculatedFieldEntityCtxIdProto, CalculatedFieldStateProto> results = new HashMap<>(); |
|
||||
try (RocksIterator iterator = db.newIterator()) { |
|
||||
for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { |
|
||||
try { |
|
||||
CalculatedFieldEntityCtxIdProto key = CalculatedFieldEntityCtxIdProto.parseFrom(iterator.key()); |
|
||||
CalculatedFieldStateProto value = CalculatedFieldStateProto.parseFrom(iterator.value()); |
|
||||
results.put(key, value); |
|
||||
} catch (Exception e) { |
|
||||
log.error("Failed to retrieve data from RocksDB", e); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
return results; |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -0,0 +1,71 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2024 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.utils; |
||||
|
|
||||
|
import lombok.SneakyThrows; |
||||
|
import org.rocksdb.Options; |
||||
|
import org.rocksdb.RocksDB; |
||||
|
import org.rocksdb.RocksIterator; |
||||
|
import org.rocksdb.WriteOptions; |
||||
|
|
||||
|
import java.nio.charset.StandardCharsets; |
||||
|
import java.nio.file.Files; |
||||
|
import java.nio.file.Path; |
||||
|
import java.util.function.BiConsumer; |
||||
|
|
||||
|
public class TbRocksDb { |
||||
|
|
||||
|
protected final String path; |
||||
|
private final WriteOptions writeOptions; |
||||
|
protected final RocksDB db; |
||||
|
|
||||
|
static { |
||||
|
RocksDB.loadLibrary(); |
||||
|
} |
||||
|
|
||||
|
public TbRocksDb(String path, Options dbOptions, WriteOptions writeOptions) throws Exception { |
||||
|
this.path = path; |
||||
|
this.writeOptions = writeOptions; |
||||
|
Files.createDirectories(Path.of(path).getParent()); |
||||
|
this.db = RocksDB.open(dbOptions, path); |
||||
|
} |
||||
|
|
||||
|
@SneakyThrows |
||||
|
public void put(String key, byte[] value) { |
||||
|
db.put(writeOptions, key.getBytes(StandardCharsets.UTF_8), value); |
||||
|
} |
||||
|
|
||||
|
public void forEach(BiConsumer<String, byte[]> processor) { |
||||
|
try (RocksIterator iterator = db.newIterator()) { |
||||
|
for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { |
||||
|
String key = new String(iterator.key(), StandardCharsets.UTF_8); |
||||
|
processor.accept(key, iterator.value()); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@SneakyThrows |
||||
|
public void delete(String key) { |
||||
|
db.delete(writeOptions, key.getBytes(StandardCharsets.UTF_8)); |
||||
|
} |
||||
|
|
||||
|
public void close() { |
||||
|
if (db != null) { |
||||
|
db.close(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue