Andrii Shvaika
7 years ago
11 changed files with 13 additions and 402 deletions
@ -1,225 +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.service.transaction; |
|
||||
|
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.springframework.beans.factory.annotation.Value; |
|
||||
import org.springframework.stereotype.Service; |
|
||||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|
||||
import org.thingsboard.rule.engine.api.RuleChainTransactionService; |
|
||||
import org.thingsboard.server.common.data.id.EntityId; |
|
||||
import org.thingsboard.server.common.msg.TbMsg; |
|
||||
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
|
||||
import org.thingsboard.server.service.executors.DbCallbackExecutorService; |
|
||||
|
|
||||
import javax.annotation.PostConstruct; |
|
||||
import javax.annotation.PreDestroy; |
|
||||
import java.util.Queue; |
|
||||
import java.util.concurrent.BlockingQueue; |
|
||||
import java.util.concurrent.Callable; |
|
||||
import java.util.concurrent.ConcurrentHashMap; |
|
||||
import java.util.concurrent.ConcurrentLinkedQueue; |
|
||||
import java.util.concurrent.ConcurrentMap; |
|
||||
import java.util.concurrent.ExecutorService; |
|
||||
import java.util.concurrent.Executors; |
|
||||
import java.util.concurrent.LinkedBlockingQueue; |
|
||||
import java.util.concurrent.TimeUnit; |
|
||||
import java.util.concurrent.locks.Lock; |
|
||||
import java.util.concurrent.locks.ReentrantLock; |
|
||||
import java.util.function.Consumer; |
|
||||
|
|
||||
@Service |
|
||||
@TbRuleEngineComponent |
|
||||
@Slf4j |
|
||||
public class BaseRuleChainTransactionService implements RuleChainTransactionService { |
|
||||
|
|
||||
private final DbCallbackExecutorService callbackExecutor; |
|
||||
|
|
||||
@Value("${actors.rule.transaction.queue_size}") |
|
||||
private int finalQueueSize; |
|
||||
@Value("${actors.rule.transaction.duration}") |
|
||||
private long duration; |
|
||||
|
|
||||
private final Lock transactionLock = new ReentrantLock(); |
|
||||
private final ConcurrentMap<EntityId, BlockingQueue<TbTransactionTask>> transactionMap = new ConcurrentHashMap<>(); |
|
||||
private final Queue<TbTransactionTask> timeoutQueue = new ConcurrentLinkedQueue<>(); |
|
||||
|
|
||||
private ExecutorService timeoutExecutor; |
|
||||
|
|
||||
public BaseRuleChainTransactionService(DbCallbackExecutorService callbackExecutor) { |
|
||||
this.callbackExecutor = callbackExecutor; |
|
||||
} |
|
||||
|
|
||||
@PostConstruct |
|
||||
public void init() { |
|
||||
timeoutExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("rule-chain-transaction")); |
|
||||
executeOnTimeout(); |
|
||||
} |
|
||||
|
|
||||
@PreDestroy |
|
||||
public void destroy() { |
|
||||
if (timeoutExecutor != null) { |
|
||||
timeoutExecutor.shutdownNow(); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
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 -> |
|
||||
new LinkedBlockingQueue<>(finalQueueSize)); |
|
||||
|
|
||||
TbTransactionTask transactionTask = new TbTransactionTask(msg, onStart, onEnd, onFailure, System.currentTimeMillis() + duration); |
|
||||
int queueSize = queue.size(); |
|
||||
if (queueSize >= finalQueueSize) { |
|
||||
log.trace("Queue has no space: {}", transactionTask); |
|
||||
executeOnFailure(transactionTask.getOnFailure(), "Queue has no space!"); |
|
||||
} else { |
|
||||
addMsgToQueues(queue, transactionTask); |
|
||||
if (queueSize == 0) { |
|
||||
executeOnSuccess(transactionTask.getOnStart(), transactionTask.getMsg()); |
|
||||
} else { |
|
||||
log.trace("Msg [{}][{}] is waiting to start transaction!", msg.getId(), msg.getType()); |
|
||||
} |
|
||||
} |
|
||||
} finally { |
|
||||
transactionLock.unlock(); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void endTransaction(TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure) { |
|
||||
//TODO 2.5
|
|
||||
// Optional<ServerAddress> address = routingService.resolveById(msg.getTransactionData().getOriginatorId());
|
|
||||
// if (address.isPresent()) {
|
|
||||
// sendTransactionEventToRemoteServer(msg, address.get());
|
|
||||
// executeOnSuccess(onSuccess, msg);
|
|
||||
// } else {
|
|
||||
endLocalTransaction(msg, onSuccess, onFailure); |
|
||||
// }
|
|
||||
} |
|
||||
|
|
||||
private void addMsgToQueues(BlockingQueue<TbTransactionTask> queue, TbTransactionTask transactionTask) { |
|
||||
queue.offer(transactionTask); |
|
||||
timeoutQueue.offer(transactionTask); |
|
||||
log.trace("Added msg to queue, size: [{}]", queue.size()); |
|
||||
} |
|
||||
|
|
||||
private void endLocalTransaction(TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure) { |
|
||||
transactionLock.lock(); |
|
||||
try { |
|
||||
BlockingQueue<TbTransactionTask> queue = transactionMap.computeIfAbsent(msg.getTransactionData().getOriginatorId(), 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()); |
|
||||
|
|
||||
executeOnSuccess(currentTransactionTask.getOnEnd(), currentTransactionTask.getMsg()); |
|
||||
executeOnSuccess(onSuccess, msg); |
|
||||
|
|
||||
TbTransactionTask nextTransactionTask = queue.peek(); |
|
||||
if (nextTransactionTask != null) { |
|
||||
executeOnSuccess(nextTransactionTask.getOnStart(), nextTransactionTask.getMsg()); |
|
||||
} |
|
||||
} else { |
|
||||
log.trace("Task has expired!"); |
|
||||
executeOnFailure(onFailure, "Task has expired!"); |
|
||||
} |
|
||||
} else { |
|
||||
log.trace("Queue is empty, previous task has expired!"); |
|
||||
executeOnFailure(onFailure, "Queue is empty, previous task has expired!"); |
|
||||
} |
|
||||
} finally { |
|
||||
transactionLock.unlock(); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
private void executeOnTimeout() { |
|
||||
timeoutExecutor.submit(() -> { |
|
||||
while (true) { |
|
||||
TbTransactionTask transactionTask = timeoutQueue.peek(); |
|
||||
if (transactionTask != null) { |
|
||||
long sleepDuration = 0L; |
|
||||
transactionLock.lock(); |
|
||||
try { |
|
||||
if (transactionTask.isCompleted()) { |
|
||||
timeoutQueue.poll(); |
|
||||
} else { |
|
||||
long expIn = transactionTask.getExpirationTime() - System.currentTimeMillis(); |
|
||||
if (expIn < 0) { |
|
||||
log.trace("Task has expired! Deleting it...[{}][{}]", transactionTask.getMsg().getId(), transactionTask.getMsg().getType()); |
|
||||
timeoutQueue.poll(); |
|
||||
executeOnFailure(transactionTask.getOnFailure(), "Task has expired!"); |
|
||||
|
|
||||
BlockingQueue<TbTransactionTask> queue = transactionMap.get(transactionTask.getMsg().getTransactionData().getOriginatorId()); |
|
||||
if (queue != null) { |
|
||||
queue.poll(); |
|
||||
TbTransactionTask nextTransactionTask = queue.peek(); |
|
||||
if (nextTransactionTask != null) { |
|
||||
executeOnSuccess(nextTransactionTask.getOnStart(), nextTransactionTask.getMsg()); |
|
||||
} |
|
||||
} |
|
||||
} else { |
|
||||
sleepDuration = Math.min(expIn, duration); |
|
||||
} |
|
||||
} |
|
||||
} finally { |
|
||||
transactionLock.unlock(); |
|
||||
} |
|
||||
if (sleepDuration > 0L) { |
|
||||
try { |
|
||||
log.trace("Task has not expired! Continue executing...[{}][{}]", transactionTask.getMsg().getId(), transactionTask.getMsg().getType()); |
|
||||
TimeUnit.MILLISECONDS.sleep(sleepDuration); |
|
||||
} catch (InterruptedException e) { |
|
||||
throw new IllegalStateException("Thread interrupted", e); |
|
||||
} |
|
||||
} |
|
||||
} else { |
|
||||
try { |
|
||||
log.trace("Queue is empty, waiting for tasks!"); |
|
||||
TimeUnit.SECONDS.sleep(1); |
|
||||
} catch (InterruptedException e) { |
|
||||
throw new IllegalStateException("Thread interrupted", e); |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
}); |
|
||||
} |
|
||||
|
|
||||
private void executeOnFailure(Consumer<Throwable> onFailure, String exception) { |
|
||||
executeCallback(() -> { |
|
||||
onFailure.accept(new RuntimeException(exception)); |
|
||||
return null; |
|
||||
}); |
|
||||
} |
|
||||
|
|
||||
private void executeOnSuccess(Consumer<TbMsg> onSuccess, TbMsg tbMsg) { |
|
||||
executeCallback(() -> { |
|
||||
onSuccess.accept(tbMsg); |
|
||||
return null; |
|
||||
}); |
|
||||
} |
|
||||
|
|
||||
private void executeCallback(Callable<Void> task) { |
|
||||
callbackExecutor.executeAsync(task); |
|
||||
} |
|
||||
|
|
||||
} |
|
||||
@ -1,44 +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.service.transaction; |
|
||||
|
|
||||
import lombok.AllArgsConstructor; |
|
||||
import lombok.Data; |
|
||||
import org.thingsboard.server.common.msg.TbMsg; |
|
||||
|
|
||||
import java.util.function.Consumer; |
|
||||
|
|
||||
@Data |
|
||||
@AllArgsConstructor |
|
||||
public final class TbTransactionTask { |
|
||||
|
|
||||
private final TbMsg msg; |
|
||||
private final Consumer<TbMsg> onStart; |
|
||||
private final Consumer<TbMsg> onEnd; |
|
||||
private final Consumer<Throwable> onFailure; |
|
||||
private final long expirationTime; |
|
||||
|
|
||||
private boolean isCompleted; |
|
||||
|
|
||||
public TbTransactionTask(TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure, long expirationTime) { |
|
||||
this.msg = msg; |
|
||||
this.onStart = onStart; |
|
||||
this.onEnd = onEnd; |
|
||||
this.onFailure = onFailure; |
|
||||
this.expirationTime = expirationTime; |
|
||||
this.isCompleted = false; |
|
||||
} |
|
||||
} |
|
||||
@ -1,30 +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.common.msg; |
|
||||
|
|
||||
import lombok.Data; |
|
||||
import org.thingsboard.server.common.data.id.EntityId; |
|
||||
|
|
||||
import java.io.Serializable; |
|
||||
import java.util.UUID; |
|
||||
|
|
||||
@Data |
|
||||
public final class TbMsgTransactionData implements Serializable { |
|
||||
|
|
||||
private final UUID transactionId; |
|
||||
private final EntityId originatorId; |
|
||||
|
|
||||
} |
|
||||
@ -1,28 +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.rule.engine.api; |
|
||||
|
|
||||
import org.thingsboard.server.common.msg.TbMsg; |
|
||||
|
|
||||
import java.util.function.Consumer; |
|
||||
|
|
||||
public interface RuleChainTransactionService { |
|
||||
|
|
||||
void beginTransaction(TbMsg msg, Consumer<TbMsg> onStart, Consumer<TbMsg> onEnd, Consumer<Throwable> onFailure); |
|
||||
|
|
||||
void endTransaction(TbMsg msg, Consumer<TbMsg> onSuccess, Consumer<Throwable> onFailure); |
|
||||
|
|
||||
} |
|
||||
Loading…
Reference in new issue