diff --git a/application/pom.xml b/application/pom.xml
index 4cbd9c3b64..86d50604e7 100644
--- a/application/pom.xml
+++ b/application/pom.xml
@@ -382,35 +382,35 @@
rocksdbjni
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-open-ai
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-azure-open-ai
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-google-ai-gemini
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-vertex-ai-gemini
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-mistral-ai
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-anthropic
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-bedrock
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-github-models
@@ -420,7 +420,7 @@
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-ollama
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
index 9fcb7425b2..1827b986fd 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
@@ -69,6 +69,7 @@ import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -82,6 +83,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
+import java.util.function.Function;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME;
@@ -94,6 +96,7 @@ import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAS
public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase implements EdgeRpcService {
private final ConcurrentMap sessions = new ConcurrentHashMap<>();
+ private final ConcurrentMap sessionsById = new ConcurrentHashMap<>();
private final ConcurrentMap sessionNewEventsLocks = new ConcurrentHashMap<>();
private final Map sessionNewEvents = new HashMap<>();
private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>();
@@ -283,6 +286,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
destroySession(session);
session.cleanUp();
sessions.remove(edgeId);
+ sessionsById.remove(session.getSessionId());
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
@@ -332,9 +336,15 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
TenantId tenantId = edge.getTenantId();
log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId);
if (sessions.containsKey(edgeId)) {
- destroySession(sessions.get(edgeId));
+ EdgeGrpcSession existing = sessions.get(edgeId);
+ if (existing != null) {
+ log.info("[{}][{}] Replacing existing session [{}] for edge [{}]", tenantId, edgeGrpcSession.getSessionId(), existing.getSessionId(), edgeId);
+ destroySession(existing);
+ sessionsById.remove(existing.getSessionId());
+ }
}
sessions.put(edgeId, edgeGrpcSession);
+ sessionsById.put(edgeGrpcSession.getSessionId(), edgeGrpcSession);
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
@@ -492,9 +502,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void onEdgeDisconnect(Edge edge, UUID sessionId) {
EdgeId edgeId = edge.getId();
log.info("[{}][{}] edge disconnected!", edgeId, sessionId);
- EdgeGrpcSession toRemove = sessions.get(edgeId);
- if (toRemove.getSessionId().equals(sessionId)) {
- toRemove = sessions.remove(edgeId);
+ EdgeGrpcSession current = sessions.get(edgeId);
+ if (current != null && current.getSessionId().equals(sessionId)) {
+ EdgeGrpcSession toRemove = sessions.remove(edgeId);
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
@@ -503,6 +513,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
newEventLock.unlock();
}
destroySession(toRemove);
+ sessionsById.remove(sessionId);
TenantId tenantId = toRemove.getEdge().getTenantId();
save(tenantId, edgeId, ACTIVITY_STATE, false);
long lastDisconnectTs = System.currentTimeMillis();
@@ -510,7 +521,18 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT);
cancelScheduleEdgeEventsCheck(edgeId);
} else {
- log.debug("[{}] edge session [{}] is not available anymore, nothing to remove. most probably this session is already outdated!", edgeId, sessionId);
+ log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId);
+ EdgeGrpcSession stale = sessionsById.remove(sessionId);
+ if (stale != null) {
+ try {
+ destroySession(stale);
+ log.info("[{}][{}] Successfully destroyed stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId);
+ } catch (Exception e) {
+ log.warn("[{}][{}] Failed to destroy stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId, e);
+ }
+ } else {
+ log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId);
+ }
}
edgeIdServiceIdCache.evict(edgeId);
}
@@ -522,6 +544,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId());
zombieSessions.add(session);
}
+ } catch (Exception e) {
+ log.warn("[{}][{}] Exception during session destroy for edge [{}] with session id [{}]",
+ session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId(), e);
}
}
@@ -631,25 +656,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void cleanupZombieSessions() {
try {
- List toRemove = new ArrayList<>();
- for (EdgeGrpcSession session : sessions.values()) {
- if (session instanceof KafkaEdgeGrpcSession kafkaSession &&
- !kafkaSession.isConnected() &&
- kafkaSession.getConsumer() != null &&
- kafkaSession.getConsumer().getConsumer() != null &&
- !kafkaSession.getConsumer().getConsumer().isStopped()) {
- toRemove.add(kafkaSession.getEdge().getId());
- }
- }
- for (EdgeId edgeId : toRemove) {
- log.info("[{}] Destroying session for edge because edge is not connected", edgeId);
- EdgeGrpcSession removed = sessions.get(edgeId);
- if (removed instanceof KafkaEdgeGrpcSession kafkaSession) {
- if (kafkaSession.destroy()) {
- sessions.remove(edgeId);
- }
- }
- }
+ tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId()));
+ tryToDestroyZombieSessions(getZombieSessions(sessionsById.values()), s -> sessionsById.remove(s.getSessionId()));
+
zombieSessions.removeIf(zombie -> {
if (zombie.destroy()) {
log.info("[{}][{}] Successfully cleaned up zombie session [{}] for edge [{}].",
@@ -665,4 +674,38 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
log.warn("Failed to cleanup kafka sessions", e);
}
}
+
+ private List getZombieSessions(Collection sessions) {
+ List result = new ArrayList<>();
+ for (EdgeGrpcSession session : sessions) {
+ if (isKafkaSessionAndZombie(session)) {
+ result.add(session);
+ }
+ }
+ return result;
+ }
+
+ private void tryToDestroyZombieSessions(List sessionsToRemove, Function removeFunc) {
+ for (EdgeGrpcSession toRemove : sessionsToRemove) {
+ log.info("[{}] Destroying session for edge because edge is not connected", toRemove.getEdge().getId());
+ if (toRemove.destroy()) {
+ removeFunc.apply(toRemove);
+ }
+ }
+ }
+
+ private boolean isKafkaSessionAndZombie(EdgeGrpcSession session) {
+ if (session instanceof KafkaEdgeGrpcSession kafkaSession) {
+ log.debug("[{}] kafkaSession.isConnected() = {}, kafkaSession.getConsumer().getConsumer().isStopped() = {}",
+ kafkaSession.getEdge().getId(),
+ kafkaSession.isConnected(),
+ kafkaSession.getConsumer() != null ? kafkaSession.getConsumer().getConsumer() != null ? kafkaSession.getConsumer().getConsumer().isStopped() : null : null);
+ return !kafkaSession.isConnected() &&
+ kafkaSession.getConsumer() != null &&
+ kafkaSession.getConsumer().getConsumer() != null &&
+ !kafkaSession.getConsumer().getConsumer().isStopped();
+ }
+ return false;
+ }
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
index d165be33d4..63669a8e3d 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
@@ -101,8 +101,21 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
@Override
public ListenableFuture processEdgeEvents() {
+ if (!isConnected() || isSyncInProgress() || isHighPriorityProcessing) {
+ log.warn("[{}][{}] Session is not ready (connected={}, syncInProgress={}, highPriority={}), skip starting edge event consumer",
+ tenantId, edge != null ? edge.getId() : null, isConnected(), isSyncInProgress(), isHighPriorityProcessing);
+ return Futures.immediateFuture(Boolean.FALSE);
+ }
if (consumer == null || (consumer.getConsumer() != null && consumer.getConsumer().isStopped())) {
try {
+ if (consumerExecutor != null && !consumerExecutor.isShutdown()) {
+ try {
+ consumerExecutor.shutdown();
+ awaitConsumerTermination();
+ } catch (Exception e) {
+ log.warn("[{}][{}] Failed to shutdown previous consumer executor", tenantId, edge.getId(), e);
+ }
+ }
this.consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-event-consumer"));
this.consumer = QueueConsumerManager.>builder()
.name("TB Edge events [" + edge.getId() + "]")
@@ -133,6 +146,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
public boolean destroy() {
try {
if (consumer != null) {
+ log.info("[{}][{}] Stopping edge event consumer...", tenantId, edge != null ? edge.getId() : null);
consumer.stop();
}
} catch (Exception e) {
@@ -141,16 +155,25 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
}
consumer = null;
try {
- if (consumerExecutor != null) {
+ if (consumerExecutor != null && !consumerExecutor.isShutdown()) {
consumerExecutor.shutdown();
+ awaitConsumerTermination();
}
} catch (Exception e) {
- log.warn("[{}][{}] Failed to shutdown consumer executor", tenantId, edge.getId(), e);
+ log.warn("[{}][{}] Failed to shutdown edge event consumer executor", tenantId, edge.getId(), e);
return false;
}
return true;
}
+ private void awaitConsumerTermination() {
+ try {
+ consumerExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS);
+ } catch (InterruptedException ie) {
+ log.warn("[{}][{}] Interrupted while awaiting consumer executor termination", tenantId, edge.getId());
+ }
+ }
+
@Override
public void cleanUp() {
String topic = topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edge.getId()).getTopic();
diff --git a/common/data/pom.xml b/common/data/pom.xml
index 779565ed8e..3afd7cf779 100644
--- a/common/data/pom.xml
+++ b/common/data/pom.xml
@@ -113,7 +113,7 @@
compile
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-core
diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
index 6542492238..a2ad982765 100644
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
@@ -105,6 +105,7 @@ public abstract class AbstractGatewaySessionHandler {
ack(msg, MqttReasonCodes.PubAck.SUCCESS);
log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName);
},
- t -> logDeviceCreationError(t, deviceName));
+ t -> processFailure(msgId, deviceName, CONNECT, ackSent, t));
}
public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional deviceProfileOpt) {
diff --git a/pom.xml b/pom.xml
index b0768966bd..4676ee3793 100755
--- a/pom.xml
+++ b/pom.xml
@@ -110,7 +110,7 @@
4.0.2
1.7.5
3.8.0
- 1.1.0
+ 1.8.0-TB
2.38.0
1.24
1.11.0
@@ -911,7 +911,7 @@
import
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j-bom
${langchain4j.version}
pom
@@ -1908,6 +1908,10 @@
central
https://repo1.maven.org/maven2/
+
+ thingsboard-repo
+ https://repo.thingsboard.io/artifactory/libs-release-public
+
spring-snapshots
Spring Snapshots
diff --git a/rule-engine/rule-engine-api/pom.xml b/rule-engine/rule-engine-api/pom.xml
index 59397d4a92..0c1bb2e362 100644
--- a/rule-engine/rule-engine-api/pom.xml
+++ b/rule-engine/rule-engine-api/pom.xml
@@ -99,7 +99,7 @@
provided
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j
diff --git a/rule-engine/rule-engine-components/pom.xml b/rule-engine/rule-engine-components/pom.xml
index 98c7fde665..598377ec86 100644
--- a/rule-engine/rule-engine-components/pom.xml
+++ b/rule-engine/rule-engine-components/pom.xml
@@ -154,7 +154,7 @@
json-path
- dev.langchain4j
+ org.thingsboard.langchain4j
langchain4j
diff --git a/ui-ngx/package.json b/ui-ngx/package.json
index 580b63a588..3403e7a3aa 100644
--- a/ui-ngx/package.json
+++ b/ui-ngx/package.json
@@ -141,6 +141,9 @@
"rollup": "4.22.4",
"@babel/core": "7.25.2",
"esbuild": "0.23.0",
- "jquery.terminal/coveralls-next/form-data": "4.0.4"
+ "jquery.terminal/coveralls-next/form-data": "4.0.4",
+ "node-forge": "1.3.3",
+ "glob": "10.5.0",
+ "path-to-regexp": "0.1.12"
}
}
diff --git a/ui-ngx/src/app/modules/home/components/ai-model/check-connectivity-dialog.component.ts b/ui-ngx/src/app/modules/home/components/ai-model/check-connectivity-dialog.component.ts
index dd2c27cb47..cd593b4291 100644
--- a/ui-ngx/src/app/modules/home/components/ai-model/check-connectivity-dialog.component.ts
+++ b/ui-ngx/src/app/modules/home/components/ai-model/check-connectivity-dialog.component.ts
@@ -56,9 +56,7 @@ export class CheckConnectivityDialogComponent extends DialogComponent implements OnInit {
+export class AlarmDetailsDialogComponent extends DialogComponent {
alarmId: string;
alarmFormGroup: UntypedFormGroup;
@@ -128,13 +128,12 @@ export class AlarmDetailsDialogComponent extends DialogComponent {
context: ctx,
showCloseButton: false,
popoverContentStyle: {padding: '16px 24px'},
- isModal: false
+ isModal: true
});
releaseNotesPanelPopover.tbComponentRef.instance.popover = releaseNotesPanelPopover;
releaseNotesPanelPopover.tbComponentRef.instance.editorContentApplied.subscribe((releaseNotes) => {
diff --git a/ui-ngx/src/app/modules/home/pages/mobile/common/editor-panel.component.ts b/ui-ngx/src/app/modules/home/pages/mobile/common/editor-panel.component.ts
index 0b5c93ff59..c177e8282b 100644
--- a/ui-ngx/src/app/modules/home/pages/mobile/common/editor-panel.component.ts
+++ b/ui-ngx/src/app/modules/home/pages/mobile/common/editor-panel.component.ts
@@ -59,9 +59,33 @@ export class EditorPanelComponent implements OnInit {
resize: false,
setup: (editor) => {
editor.on('PostRender', function() {
- const container = editor.getContainer().closest('.tb-popover-content');
- const uiContainer = document.querySelector('.tox.tox-tinymce-aux');
- container.parentNode.appendChild(uiContainer);
+ const container = document.querySelector('.tox.tox-tinymce-aux');
+ const styleSheet = document.createElement('style');
+ styleSheet.innerText = `
+ .tox-tiered-menu .tox-menu {
+ width: fit-content;
+ max-width: min(80%, 440px);
+ @media screen and (max-width: 510px) {
+ max-width: calc(100% - 64px);
+ }
+ media screen and (min-width: 511px) and (max-width: 548px) {
+ max-width: calc(100% - 84px);
+ }
+ media screen and (min-width: 549px) and (max-width: 599px) {
+ max-width: calc(100% - 104px);
+ }
+ }
+ .tox-tiered-menu .tox-menu .tox-collection__item-label {
+ word-break: normal;
+ }
+ @media screen and (max-width: 890px) {
+ .tox-tiered-menu > .tox-collection--list:not(:first-child) {
+ left: auto !important;
+ right: 0 !important;
+ }
+ }
+ `;
+ container.prepend(styleSheet);
});
},
relative_urls: false,
diff --git a/ui-ngx/src/app/modules/home/pages/notification/recipient/recipient-notification-dialog.component.html b/ui-ngx/src/app/modules/home/pages/notification/recipient/recipient-notification-dialog.component.html
index 95715bef87..560db89dbc 100644
--- a/ui-ngx/src/app/modules/home/pages/notification/recipient/recipient-notification-dialog.component.html
+++ b/ui-ngx/src/app/modules/home/pages/notification/recipient/recipient-notification-dialog.component.html
@@ -70,6 +70,7 @@
{
+ editor.on('PostRender', function() {
+ const container = document.querySelector('.tox.tox-tinymce-aux');
+ const styleSheet = document.createElement('style');
+ styleSheet.innerText = `
+ .tox-tiered-menu .tox-menu {
+ width: fit-content;
+ max-width: min(80%, 440px);
+ @media screen and (max-width: 510px) {
+ max-width: calc(100% - 64px);
+ }
+ media screen and (min-width: 511px) and (max-width: 548px) {
+ max-width: calc(100% - 84px);
+ }
+ media screen and (min-width: 549px) and (max-width: 599px) {
+ max-width: calc(100% - 104px);
+ }
+ }
+ .tox-tiered-menu .tox-menu .tox-collection__item-label {
+ word-break: normal;
+ }
+ @media screen and (max-width: 890px) {
+ .tox-tiered-menu > .tox-collection--list:not(:first-child) {
+ left: auto !important;
+ right: 0 !important;
+ }
+ }
+ `;
+ container.prepend(styleSheet);
+ });
+ },
relative_urls: false,
urlconverter_callback: (url) => url
};
diff --git a/ui-ngx/src/app/shared/models/ai-model.models.ts b/ui-ngx/src/app/shared/models/ai-model.models.ts
index d3b70f9a74..410f7ad900 100644
--- a/ui-ngx/src/app/shared/models/ai-model.models.ts
+++ b/ui-ngx/src/app/shared/models/ai-model.models.ts
@@ -110,6 +110,7 @@ export const AiModelMap = new Map