dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

EdgeGrpcSession - move check of empty high priority msgs into submethod

Browse Source
This commit is contained in:
Volodymyr Babak 2024-12-23 20:46:07 +02:00
parent 509ff28f2e
commit d5e973283c
1 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -292,9 +292,7 @@ public abstract class EdgeGrpcSession implements Closeable {
protected void processEdgeEvents(EdgeEventFetcher fetcher, PageLink pageLink, SettableFuture<Pair<Long, Long>> result) { protected void processEdgeEvents(EdgeEventFetcher fetcher, PageLink pageLink, SettableFuture<Pair<Long, Long>> result) {
try { try {
if (!highPriorityQueue.isEmpty()) {
processHighPriorityEvents(); processHighPriorityEvents();
}
PageData<EdgeEvent> pageData = fetcher.fetchEdgeEvents(edge.getTenantId(), edge, pageLink); PageData<EdgeEvent> pageData = fetcher.fetchEdgeEvents(edge.getTenantId(), edge, pageLink);
if (isConnected() && !pageData.getData().isEmpty()) { if (isConnected() && !pageData.getData().isEmpty()) {
log.trace("[{}][{}][{}] event(s) are going to be processed.", tenantId, sessionId, pageData.getData().size()); log.trace("[{}][{}][{}] event(s) are going to be processed.", tenantId, sessionId, pageData.getData().size());
@ -537,6 +535,9 @@ public abstract class EdgeGrpcSession implements Closeable {
public void processHighPriorityEvents() { public void processHighPriorityEvents() {
try { try {
if (isConnected() && isSyncCompleted()) { if (isConnected() && isSyncCompleted()) {
if (highPriorityQueue.isEmpty()) {
return;
}
List<EdgeEvent> highPriorityEvents = new ArrayList<>(); List<EdgeEvent> highPriorityEvents = new ArrayList<>();
EdgeEvent event; EdgeEvent event;
while ((event = highPriorityQueue.poll()) != null) { while ((event = highPriorityQueue.poll()) != null) {
@ -606,9 +607,14 @@ public abstract class EdgeGrpcSession implements Closeable {
} }
}, ctx.getGrpcCallbackExecutorService()); }, ctx.getGrpcCallbackExecutorService());
} else { } else {
log.trace("[{}][{}] edge is not connected or sync is not completed. Skipping iteration", tenantId, sessionId); if (!isSyncCompleted()) {
log.trace("[{}][{}] edge sync is not completed yet. Skipping iteration", tenantId, sessionId);
result.set(Boolean.TRUE);
} else {
log.trace("[{}][{}] edge is not connected. Skipping iteration", tenantId, sessionId);
result.set(null); result.set(null);
} }
}
return result; return result;
} }