tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

improved transaction service in cluster mode

pull/1301/head
Dima Landiak 8 years ago
parent
138c049c89
commit
f980b2b6cf
8 changed files with 85 additions and 129 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 122
      application/src/main/java/org/thingsboard/server/service/transaction/BaseRuleChainTransactionService.java
  2. 12
      application/src/main/proto/cluster.proto
  3. 4
      application/src/main/resources/thingsboard.yml
  4. 4
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleChainTransactionService.java
  5. 28
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbTransactionBeginNode.java
  6. 32
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbTransactionBeginNodeConfiguration.java
  7. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbTransactionEndNode.java
  8. 6
      rule-engine/rule-engine-components/src/main/resources/public/static/rulenode/rulenode-core-config.js

122
application/src/main/java/org/thingsboard/server/service/transaction/BaseRuleChainTransactionService.java
View File

@ -21,10 +21,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.thingsboard.rule.engine.api.RuleChainTransactionService;
import org.thingsboard.rule.engine.api.TbContext;
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.msg.TbMsg;
import org.thingsboard.server.common.msg.cluster.ServerAddress;
import org.thingsboard.server.gen.cluster.ClusterAPIProtos;
@ -88,7 +86,7 @@ public class BaseRuleChainTransactionService implements RuleChainTransactionServ
}
@Override
public void beginTransaction(TbContext ctx, TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure) {
public void beginTransaction(TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure) {
transactionLock.lock();
try {
BlockingQueue<TbTransactionTask> queue = transactionMap.computeIfAbsent(msg.getTransactionData().getOriginatorId(), id ->
@ -111,49 +109,71 @@ public class BaseRuleChainTransactionService implements RuleChainTransactionServ
}
}
@Override
public void endTransaction(TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure) {
EntityId originatorId = msg.getTransactionData().getOriginatorId();
UUID transactionId = msg.getTransactionData().getTransactionId();
Optional<ServerAddress> address = routingService.resolveById(originatorId);
if (address.isPresent()) {
sendTransactionEventToRemoteServer(originatorId, transactionId, address.get());
executeOnSuccess(onSuccess, msg);
} else {
endLocalTransaction(transactionId, originatorId, onSuccess, onFailure);
}
}
@Override
public void onRemoteTransactionMsg(ServerAddress serverAddress, byte[] data) {
ClusterAPIProtos.TransactionEndServiceMsgProto proto;
try {
proto = ClusterAPIProtos.TransactionEndServiceMsgProto.parseFrom(data);
} catch (InvalidProtocolBufferException e) {
throw new RuntimeException(e);
}
EntityId originatorId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getOriginatorIdMSB(), proto.getOriginatorIdLSB()));
UUID transactionId = new UUID(proto.getTransactionIdMSB(), proto.getTransactionIdLSB());
endLocalTransaction(transactionId, originatorId, msg -> {
}, error -> {
});
}
private void addMsgToQueues(BlockingQueue<TbTransactionTask> queue, TbTransactionTask transactionTask) {
queue.offer(transactionTask);
timeoutQueue.offer(transactionTask);
log.trace("Added msg to queue, size: [{}]", queue.size());
}
@Override
public void endTransaction(TbContext ctx, TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure) {
EntityId originatorId = msg.getTransactionData().getOriginatorId();
if (onRemoteTransactionEndSync(ctx.getTenantId(), originatorId)) {
executeOnSuccess(onSuccess, msg);
} else {
transactionLock.lock();
try {
BlockingQueue<TbTransactionTask> queue = transactionMap.computeIfAbsent(originatorId, id ->
new LinkedBlockingQueue<>(finalQueueSize));
private void endLocalTransaction(UUID transactionId, EntityId originatorId, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure) {
transactionLock.lock();
try {
BlockingQueue<TbTransactionTask> queue = transactionMap.computeIfAbsent(originatorId, id ->
new LinkedBlockingQueue<>(finalQueueSize));
TbTransactionTask currentTransactionTask = queue.peek();
if (currentTransactionTask != null) {
if (currentTransactionTask.getMsg().getTransactionData().getTransactionId().equals(msg.getTransactionData().getTransactionId())) {
currentTransactionTask.setCompleted(true);
queue.poll();
log.trace("Removed msg from queue, size [{}]", queue.size());
TbTransactionTask currentTransactionTask = queue.peek();
if (currentTransactionTask != null) {
if (currentTransactionTask.getMsg().getTransactionData().getTransactionId().equals(transactionId)) {
currentTransactionTask.setCompleted(true);
queue.poll();
log.trace("Removed msg from queue, size [{}]", queue.size());
executeOnSuccess(currentTransactionTask.getOnEnd(), currentTransactionTask.getMsg());
executeOnSuccess(onSuccess, currentTransactionTask.getMsg());
executeOnSuccess(currentTransactionTask.getOnEnd(), currentTransactionTask.getMsg());
executeOnSuccess(onSuccess, currentTransactionTask.getMsg());
TbTransactionTask nextTransactionTask = queue.peek();
if (nextTransactionTask != null) {
executeOnSuccess(nextTransactionTask.getOnStart(), nextTransactionTask.getMsg());
}
} else {
log.trace("Task has expired!");
executeOnFailure(onFailure, "Task has expired!");
TbTransactionTask nextTransactionTask = queue.peek();
if (nextTransactionTask != null) {
executeOnSuccess(nextTransactionTask.getOnStart(), nextTransactionTask.getMsg());
}
} else {
log.trace("Queue is empty, previous task has expired!");
executeOnFailure(onFailure, "Queue is empty, previous task has expired!");
log.trace("Task has expired!");
executeOnFailure(onFailure, "Task has expired!");
}
} finally {
transactionLock.unlock();
} else {
log.trace("Queue is empty, previous task has expired!");
executeOnFailure(onFailure, "Queue is empty, previous task has expired!");
}
} finally {
transactionLock.unlock();
}
}
@ -222,40 +242,14 @@ public class BaseRuleChainTransactionService implements RuleChainTransactionServ
callbackExecutor.executeAsync(task);
}
@Override
public void onRemoteTransactionMsg(ServerAddress serverAddress, byte[] data) {
ClusterAPIProtos.TransactionServiceMsgProto proto;
try {
proto = ClusterAPIProtos.TransactionServiceMsgProto.parseFrom(data);
} catch (InvalidProtocolBufferException e) {
throw new RuntimeException(e);
}
TenantId tenantId = new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB()));
EntityId entityId = EntityIdFactory.getByTypeAndUuid(proto.getEntityType(), new UUID(proto.getOriginatorIdMSB(), proto.getOriginatorIdLSB()));
onTransactionEnd(tenantId, entityId);
}
private void onTransactionEnd(TenantId tenantId, EntityId entityId) {
callbackExecutor.executeAsync(() -> onRemoteTransactionEndSync(tenantId, entityId));
}
private boolean onRemoteTransactionEndSync(TenantId tenantId, EntityId entityId) {
Optional<ServerAddress> address = routingService.resolveById(entityId);
if (address.isPresent()) {
sendTransactionEvent(tenantId, entityId, address.get());
return true;
}
return false;
}
private void sendTransactionEvent(TenantId tenantId, EntityId entityId, ServerAddress address) {
log.trace("[{}][{}] Originator is monitored on other server: {}", tenantId, entityId, address);
ClusterAPIProtos.TransactionServiceMsgProto.Builder builder = ClusterAPIProtos.TransactionServiceMsgProto.newBuilder();
builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
private void sendTransactionEventToRemoteServer(EntityId entityId, UUID transactionId, ServerAddress address) {
log.trace("[{}][{}] Originator is monitored on other server: {}", entityId, transactionId, address);
ClusterAPIProtos.TransactionEndServiceMsgProto.Builder builder = ClusterAPIProtos.TransactionEndServiceMsgProto.newBuilder();
builder.setEntityType(entityId.getEntityType().name());
builder.setOriginatorIdMSB(entityId.getId().getMostSignificantBits());
builder.setOriginatorIdLSB(entityId.getId().getLeastSignificantBits());
builder.setTransactionIdMSB(transactionId.getMostSignificantBits());
builder.setTransactionIdLSB(transactionId.getLeastSignificantBits());
clusterRpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TRANSACTION_SERVICE_MESSAGE, builder.build().toByteArray());
}
}

12
application/src/main/proto/cluster.proto
View File

@ -144,10 +144,10 @@ message DeviceStateServiceMsgProto {
bool deleted = 7;
}
message TransactionServiceMsgProto {
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
string entityType = 3;
int64 originatorIdMSB = 4;
int64 originatorIdLSB = 5;
message TransactionEndServiceMsgProto {
string entityType = 1;
int64 originatorIdMSB = 2;
int64 originatorIdLSB = 3;
int64 transactionIdMSB = 4;
int64 transactionIdLSB = 5;
}

4
application/src/main/resources/thingsboard.yml
View File

@ -215,9 +215,9 @@ actors:
error_persist_frequency: "${ACTORS_RULE_NODE_ERROR_FREQUENCY:3000}"
transaction:
# Size of queues which store messages for transaction rule nodes
queue_size: "${ACTORS_RULE_TRANSACTION_QUEUE_SIZE:10}"
queue_size: "${ACTORS_RULE_TRANSACTION_QUEUE_SIZE:20}"
# Time in milliseconds for transaction to complete
duration: "${ACTORS_RULE_TRANSACTION_DURATION:10000}"
duration: "${ACTORS_RULE_TRANSACTION_DURATION:15000}"
statistics:
# Enable/disable actor statistics
enabled: "${ACTORS_STATISTICS_ENABLED:true}"

4
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleChainTransactionService.java
View File

@ -22,9 +22,9 @@ import java.util.function.Consumer;
public interface RuleChainTransactionService {
void beginTransaction(TbContext ctx, TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure);
void beginTransaction(TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure);
void endTransaction(TbContext ctx, TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure);
void endTransaction(TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure);
void onRemoteTransactionMsg(ServerAddress serverAddress, byte[] bytes);

28
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbTransactionBeginNode.java
View File

@ -16,13 +16,13 @@
package org.thingsboard.rule.engine.transaction;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.EmptyNodeConfiguration;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgDataType;
@ -36,38 +36,32 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
@RuleNode(
type = ComponentType.ACTION,
name = "transaction start",
configClazz = TbTransactionBeginNodeConfiguration.class,
nodeDescription = "Something",
nodeDetails = "Something more",
configClazz = EmptyNodeConfiguration.class,
nodeDescription = "",
nodeDetails = "",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbActionNodeTransactionBeginConfig")
configDirective = "tbNodeEmptyConfig")
public class TbTransactionBeginNode implements TbNode {
private TbTransactionBeginNodeConfiguration config;
private EmptyNodeConfiguration config;
@Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
this.config = TbNodeUtils.convert(configuration, TbTransactionBeginNodeConfiguration.class);
this.config = TbNodeUtils.convert(configuration, EmptyNodeConfiguration.class);
}
@Override
public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
log.trace("Msg enters transaction - [{}][{}]", msg.getId(), msg.getType());
EntityId entityId;
if (config.getTransactionEntity().equals("Originator")) {
entityId = msg.getOriginator();
} else {
entityId = ctx.getTenantId();
}
TbMsgTransactionData transactionData = new TbMsgTransactionData(msg.getId(), entityId);
TbMsgTransactionData transactionData = new TbMsgTransactionData(msg.getId(), msg.getOriginator());
TbMsg tbMsg = new TbMsg(msg.getId(), msg.getType(), msg.getOriginator(), msg.getMetaData(), TbMsgDataType.JSON,
msg.getData(), transactionData, msg.getRuleChainId(), msg.getRuleNodeId(), msg.getClusterPartition());
ctx.getRuleChainTransactionService().beginTransaction(ctx, tbMsg, startMsg -> {
log.trace("Transaction starting... [{}][{}]", startMsg.getId(), startMsg.getType());
ctx.getRuleChainTransactionService().beginTransaction(tbMsg, startMsg -> {
log.trace("Transaction starting...[{}][{}]", startMsg.getId(), startMsg.getType());
ctx.tellNext(startMsg, SUCCESS);
}, endMsg -> log.trace("Transaction ended successfully... [{}][{}]", endMsg.getId(), endMsg.getType()),
}, endMsg -> log.trace("Transaction ended successfully...[{}][{}]", endMsg.getId(), endMsg.getType()),
throwable -> {
log.error("Transaction failed! [{}][{}]", tbMsg.getId(), tbMsg.getType(), throwable);
ctx.tellFailure(tbMsg, throwable);

32
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbTransactionBeginNodeConfiguration.java
View File

@ -1,32 +0,0 @@
/**
* Copyright © 2016-2018 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.rule.engine.transaction;
import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;
@Data
public class TbTransactionBeginNodeConfiguration implements NodeConfiguration<TbTransactionBeginNodeConfiguration> {
private String transactionEntity;
@Override
public TbTransactionBeginNodeConfiguration defaultConfiguration() {
TbTransactionBeginNodeConfiguration configuration = new TbTransactionBeginNodeConfiguration();
configuration.setTransactionEntity("Originator");
return configuration;
}
}

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transaction/TbTransactionEndNode.java
View File

@ -35,8 +35,8 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
type = ComponentType.ACTION,
name = "transaction end",
configClazz = EmptyNodeConfiguration.class,
nodeDescription = "Something",
nodeDetails = "Something more",
nodeDescription = "",
nodeDetails = "",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = ("tbNodeEmptyConfig")
)
@ -51,7 +51,7 @@ public class TbTransactionEndNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
ctx.getRuleChainTransactionService().endTransaction(ctx, msg,
ctx.getRuleChainTransactionService().endTransaction(msg,
successMsg -> ctx.tellNext(successMsg, SUCCESS),
throwable -> ctx.tellFailure(msg, throwable));
log.trace("Msg left transaction - [{}][{}]", msg.getId(), msg.getType());

6
rule-engine/rule-engine-components/src/main/resources/public/static/rulenode/rulenode-core-config.js
View File

File diff suppressed because one or more lines are too long
Write Preview
Loading…
Cancel
Save