Viacheslav Klimov
3 years ago
committed by
GitHub
GitHub
262 changed files with 7187 additions and 1606 deletions
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@ -0,0 +1,129 @@ |
|||||
|
/** |
||||
|
* 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.notification.channels; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.google.common.base.Strings; |
||||
|
import com.google.firebase.messaging.FirebaseMessagingException; |
||||
|
import com.google.firebase.messaging.MessagingErrorCode; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.rule.engine.api.notification.FirebaseService; |
||||
|
import org.thingsboard.server.common.data.User; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
||||
|
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
||||
|
import org.thingsboard.server.common.data.notification.settings.MobileAppNotificationDeliveryMethodConfig; |
||||
|
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
||||
|
import org.thingsboard.server.common.data.notification.template.MobileAppDeliveryMethodNotificationTemplate; |
||||
|
import org.thingsboard.server.dao.notification.NotificationSettingsService; |
||||
|
import org.thingsboard.server.dao.user.UserService; |
||||
|
import org.thingsboard.server.service.notification.NotificationProcessingContext; |
||||
|
|
||||
|
import java.util.HashMap; |
||||
|
import java.util.HashSet; |
||||
|
import java.util.Map; |
||||
|
import java.util.Optional; |
||||
|
import java.util.Set; |
||||
|
|
||||
|
@Component |
||||
|
@RequiredArgsConstructor |
||||
|
@Slf4j |
||||
|
public class MobileAppNotificationChannel implements NotificationChannel<User, MobileAppDeliveryMethodNotificationTemplate> { |
||||
|
|
||||
|
private final FirebaseService firebaseService; |
||||
|
private final UserService userService; |
||||
|
private final NotificationSettingsService notificationSettingsService; |
||||
|
|
||||
|
@Override |
||||
|
public void sendNotification(User recipient, MobileAppDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception { |
||||
|
var mobileSessions = userService.findMobileSessions(recipient.getTenantId(), recipient.getId()); |
||||
|
if (mobileSessions.isEmpty()) { |
||||
|
throw new IllegalArgumentException("User doesn't use the mobile app"); |
||||
|
} |
||||
|
|
||||
|
MobileAppNotificationDeliveryMethodConfig config = ctx.getDeliveryMethodConfig(NotificationDeliveryMethod.MOBILE_APP); |
||||
|
String credentials = config.getFirebaseServiceAccountCredentials(); |
||||
|
Set<String> validTokens = new HashSet<>(mobileSessions.keySet()); |
||||
|
|
||||
|
String subject = processedTemplate.getSubject(); |
||||
|
String body = processedTemplate.getBody(); |
||||
|
Map<String, String> data = getNotificationData(processedTemplate, ctx); |
||||
|
for (String token : mobileSessions.keySet()) { |
||||
|
try { |
||||
|
firebaseService.sendMessage(ctx.getTenantId(), credentials, token, subject, body, data); |
||||
|
} catch (FirebaseMessagingException e) { |
||||
|
MessagingErrorCode errorCode = e.getMessagingErrorCode(); |
||||
|
if (errorCode == MessagingErrorCode.UNREGISTERED || errorCode == MessagingErrorCode.INVALID_ARGUMENT) { |
||||
|
validTokens.remove(token); |
||||
|
userService.removeMobileSession(recipient.getTenantId(), token); |
||||
|
continue; |
||||
|
} |
||||
|
throw new RuntimeException("Failed to send message via FCM: " + e.getMessage(), e); |
||||
|
} |
||||
|
} |
||||
|
if (validTokens.isEmpty()) { |
||||
|
throw new IllegalArgumentException("User doesn't use the mobile app"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private Map<String, String> getNotificationData(MobileAppDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) { |
||||
|
Map<String, String> data = Optional.ofNullable(processedTemplate.getAdditionalConfig()) |
||||
|
.filter(JsonNode::isObject).map(JacksonUtil::toFlatMap).orElseGet(HashMap::new); |
||||
|
NotificationInfo info = ctx.getRequest().getInfo(); |
||||
|
if (info == null) { |
||||
|
return data; |
||||
|
} |
||||
|
Optional.ofNullable(info.getStateEntityId()).ifPresent(stateEntityId -> { |
||||
|
data.put("stateEntityId", stateEntityId.getId().toString()); |
||||
|
data.put("stateEntityType", stateEntityId.getEntityType().name()); |
||||
|
if (!"true".equals(data.get("onClick.enabled")) && info.getDashboardId() != null) { |
||||
|
data.put("onClick.enabled", "true"); |
||||
|
data.put("onClick.linkType", "DASHBOARD"); |
||||
|
data.put("onClick.setEntityIdInState", "true"); |
||||
|
data.put("onClick.dashboardId", info.getDashboardId().toString()); |
||||
|
} |
||||
|
}); |
||||
|
data.put("notificationType", ctx.getNotificationType().name()); |
||||
|
switch (ctx.getNotificationType()) { |
||||
|
case ALARM: |
||||
|
case ALARM_ASSIGNMENT: |
||||
|
case ALARM_COMMENT: |
||||
|
info.getTemplateData().forEach((key, value) -> { |
||||
|
data.put("info." + key, value); |
||||
|
}); |
||||
|
break; |
||||
|
} |
||||
|
data.replaceAll((key, value) -> Strings.nullToEmpty(value)); |
||||
|
return data; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void check(TenantId tenantId) throws Exception { |
||||
|
NotificationSettings systemSettings = notificationSettingsService.findNotificationSettings(TenantId.SYS_TENANT_ID); |
||||
|
if (!systemSettings.getDeliveryMethodsConfigs().containsKey(NotificationDeliveryMethod.MOBILE_APP)) { |
||||
|
throw new RuntimeException("Push-notifications to mobile are not configured"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public NotificationDeliveryMethod getDeliveryMethod() { |
||||
|
return NotificationDeliveryMethod.MOBILE_APP; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,133 @@ |
|||||
|
/** |
||||
|
* 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.notification.provider; |
||||
|
|
||||
|
import com.github.benmanes.caffeine.cache.Cache; |
||||
|
import com.github.benmanes.caffeine.cache.Caffeine; |
||||
|
import com.github.benmanes.caffeine.cache.RemovalCause; |
||||
|
import com.google.auth.oauth2.GoogleCredentials; |
||||
|
import com.google.firebase.FirebaseApp; |
||||
|
import com.google.firebase.FirebaseOptions; |
||||
|
import com.google.firebase.messaging.AndroidConfig; |
||||
|
import com.google.firebase.messaging.FirebaseMessaging; |
||||
|
import com.google.firebase.messaging.FirebaseMessagingException; |
||||
|
import com.google.firebase.messaging.Message; |
||||
|
import com.google.firebase.messaging.Notification; |
||||
|
import lombok.Getter; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.io.IOUtils; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.rule.engine.api.notification.FirebaseService; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
import java.io.IOException; |
||||
|
import java.nio.charset.StandardCharsets; |
||||
|
import java.util.Map; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@Service |
||||
|
@Slf4j |
||||
|
public class DefaultFirebaseService implements FirebaseService { |
||||
|
|
||||
|
private final Cache<String, FirebaseContext> contexts = Caffeine.newBuilder() |
||||
|
.expireAfterAccess(1, TimeUnit.DAYS) |
||||
|
.<String, FirebaseContext>removalListener((key, context, cause) -> { |
||||
|
if (cause == RemovalCause.EXPIRED && context != null) { |
||||
|
context.destroy(); |
||||
|
} |
||||
|
}) |
||||
|
.build(); |
||||
|
|
||||
|
@Override |
||||
|
public void sendMessage(TenantId tenantId, String credentials, String fcmToken, String title, String body, Map<String, String> data) throws FirebaseMessagingException { |
||||
|
FirebaseContext firebaseContext = contexts.asMap().compute(tenantId.toString(), (key, context) -> { |
||||
|
if (context == null) { |
||||
|
return new FirebaseContext(key, credentials); |
||||
|
} else { |
||||
|
context.check(credentials); |
||||
|
return context; |
||||
|
} |
||||
|
}); |
||||
|
|
||||
|
Message message = Message.builder() |
||||
|
.setToken(fcmToken) |
||||
|
.setNotification(Notification.builder() |
||||
|
.setTitle(title) |
||||
|
.setBody(body) |
||||
|
.build()) |
||||
|
.setAndroidConfig(AndroidConfig.builder() |
||||
|
.setPriority(AndroidConfig.Priority.HIGH) |
||||
|
.build()) |
||||
|
.putAllData(data) |
||||
|
.build(); |
||||
|
firebaseContext.getMessaging().send(message); |
||||
|
log.trace("[{}] Sent message for FCM token {}", tenantId, fcmToken); |
||||
|
} |
||||
|
|
||||
|
public static class FirebaseContext { |
||||
|
private final String key; |
||||
|
private String credentials; |
||||
|
private FirebaseApp app; |
||||
|
@Getter |
||||
|
private FirebaseMessaging messaging; |
||||
|
|
||||
|
public FirebaseContext(String key, String credentials) { |
||||
|
this.key = key; |
||||
|
this.credentials = credentials; |
||||
|
init(); |
||||
|
} |
||||
|
|
||||
|
private void init() { |
||||
|
FirebaseOptions options; |
||||
|
try { |
||||
|
options = FirebaseOptions.builder() |
||||
|
.setCredentials(GoogleCredentials.fromStream(IOUtils.toInputStream(credentials, StandardCharsets.UTF_8))) |
||||
|
.build(); |
||||
|
} catch (IOException e) { |
||||
|
throw new RuntimeException("Failed to process service account credentials: " + e.getMessage(), e); |
||||
|
} |
||||
|
try { |
||||
|
app = FirebaseApp.initializeApp(options, key); |
||||
|
} catch (IllegalStateException alreadyExists) { // should never normally happen
|
||||
|
app = FirebaseApp.getInstance(key); |
||||
|
} |
||||
|
try { |
||||
|
messaging = FirebaseMessaging.getInstance(app); |
||||
|
} catch (IllegalStateException alreadyExists) { // should never normally happen
|
||||
|
messaging = FirebaseMessaging.getInstance(app); |
||||
|
} |
||||
|
log.debug("[{}] Initialized new FirebaseContext", key); |
||||
|
} |
||||
|
|
||||
|
public void check(String credentials) { |
||||
|
if (!this.credentials.equals(credentials)) { |
||||
|
destroy(); |
||||
|
this.credentials = credentials; |
||||
|
init(); |
||||
|
} else if (app == null || messaging == null) { |
||||
|
throw new IllegalStateException("Firebase app couldn't be initialized"); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void destroy() { |
||||
|
app.delete(); |
||||
|
app = null; |
||||
|
messaging = null; |
||||
|
log.debug("[{}] Destroyed FirebaseContext", key); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,196 @@ |
|||||
|
/** |
||||
|
* 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.state; |
||||
|
|
||||
|
import lombok.Getter; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.rule.engine.api.RuleEngineDeviceStateManager; |
||||
|
import org.thingsboard.server.cluster.TbClusterService; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.queue.common.SimpleTbQueueCallback; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
||||
|
import org.thingsboard.server.queue.util.TbRuleEngineComponent; |
||||
|
|
||||
|
import java.util.Optional; |
||||
|
import java.util.UUID; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Service |
||||
|
@TbRuleEngineComponent |
||||
|
public class DefaultRuleEngineDeviceStateManager implements RuleEngineDeviceStateManager { |
||||
|
|
||||
|
private final TbServiceInfoProvider serviceInfoProvider; |
||||
|
private final PartitionService partitionService; |
||||
|
|
||||
|
private final Optional<DeviceStateService> deviceStateService; |
||||
|
private final TbClusterService clusterService; |
||||
|
|
||||
|
public DefaultRuleEngineDeviceStateManager( |
||||
|
TbServiceInfoProvider serviceInfoProvider, PartitionService partitionService, |
||||
|
Optional<DeviceStateService> deviceStateServiceOptional, TbClusterService clusterService |
||||
|
) { |
||||
|
this.serviceInfoProvider = serviceInfoProvider; |
||||
|
this.partitionService = partitionService; |
||||
|
this.deviceStateService = deviceStateServiceOptional; |
||||
|
this.clusterService = clusterService; |
||||
|
} |
||||
|
|
||||
|
@Getter |
||||
|
private abstract static class ConnectivityEventInfo { |
||||
|
|
||||
|
private final TenantId tenantId; |
||||
|
private final DeviceId deviceId; |
||||
|
private final long eventTime; |
||||
|
|
||||
|
private ConnectivityEventInfo(TenantId tenantId, DeviceId deviceId, long eventTime) { |
||||
|
this.tenantId = tenantId; |
||||
|
this.deviceId = deviceId; |
||||
|
this.eventTime = eventTime; |
||||
|
} |
||||
|
|
||||
|
abstract void forwardToLocalService(); |
||||
|
|
||||
|
abstract TransportProtos.ToCoreMsg toQueueMsg(); |
||||
|
|
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceConnect(TenantId tenantId, DeviceId deviceId, long connectTime, TbCallback callback) { |
||||
|
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, connectTime) { |
||||
|
@Override |
||||
|
void forwardToLocalService() { |
||||
|
deviceStateService.ifPresent(service -> service.onDeviceConnect(tenantId, deviceId, connectTime)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
TransportProtos.ToCoreMsg toQueueMsg() { |
||||
|
var deviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(connectTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceConnectMsg(deviceConnectMsg) |
||||
|
.build(); |
||||
|
} |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceActivity(TenantId tenantId, DeviceId deviceId, long activityTime, TbCallback callback) { |
||||
|
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, activityTime) { |
||||
|
@Override |
||||
|
void forwardToLocalService() { |
||||
|
deviceStateService.ifPresent(service -> service.onDeviceActivity(tenantId, deviceId, activityTime)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
TransportProtos.ToCoreMsg toQueueMsg() { |
||||
|
var deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(activityTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceActivityMsg(deviceActivityMsg) |
||||
|
.build(); |
||||
|
} |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceDisconnect(TenantId tenantId, DeviceId deviceId, long disconnectTime, TbCallback callback) { |
||||
|
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, disconnectTime) { |
||||
|
@Override |
||||
|
void forwardToLocalService() { |
||||
|
deviceStateService.ifPresent(service -> service.onDeviceDisconnect(tenantId, deviceId, disconnectTime)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
TransportProtos.ToCoreMsg toQueueMsg() { |
||||
|
var deviceDisconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(disconnectTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceDisconnectMsg(deviceDisconnectMsg) |
||||
|
.build(); |
||||
|
} |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onDeviceInactivity(TenantId tenantId, DeviceId deviceId, long inactivityTime, TbCallback callback) { |
||||
|
routeEvent(new ConnectivityEventInfo(tenantId, deviceId, inactivityTime) { |
||||
|
@Override |
||||
|
void forwardToLocalService() { |
||||
|
deviceStateService.ifPresent(service -> service.onDeviceInactivity(tenantId, deviceId, inactivityTime)); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
TransportProtos.ToCoreMsg toQueueMsg() { |
||||
|
var deviceInactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(inactivityTime) |
||||
|
.build(); |
||||
|
return TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceInactivityMsg(deviceInactivityMsg) |
||||
|
.build(); |
||||
|
} |
||||
|
}, callback); |
||||
|
} |
||||
|
|
||||
|
private void routeEvent(ConnectivityEventInfo eventInfo, TbCallback callback) { |
||||
|
var tenantId = eventInfo.getTenantId(); |
||||
|
var deviceId = eventInfo.getDeviceId(); |
||||
|
long eventTime = eventInfo.getEventTime(); |
||||
|
|
||||
|
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId); |
||||
|
if (serviceInfoProvider.isService(ServiceType.TB_CORE) && tpi.isMyPartition() && deviceStateService.isPresent()) { |
||||
|
log.debug("[{}][{}] Forwarding device connectivity event to local service. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime); |
||||
|
try { |
||||
|
eventInfo.forwardToLocalService(); |
||||
|
} catch (Exception e) { |
||||
|
log.error("[{}][{}] Failed to process device connectivity event. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime, e); |
||||
|
callback.onFailure(e); |
||||
|
return; |
||||
|
} |
||||
|
callback.onSuccess(); |
||||
|
} else { |
||||
|
TransportProtos.ToCoreMsg msg = eventInfo.toQueueMsg(); |
||||
|
log.debug("[{}][{}] Sending device connectivity message to core. Event time: [{}].", tenantId.getId(), deviceId.getId(), eventTime); |
||||
|
clusterService.pushMsgToCore(tpi, UUID.randomUUID(), msg, new SimpleTbQueueCallback(__ -> callback.onSuccess(), callback::onFailure)); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,38 @@ |
|||||
|
/** |
||||
|
* 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.ws.notification.sub; |
||||
|
|
||||
|
|
||||
|
import lombok.Getter; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.service.subscription.TbSubscription; |
||||
|
import org.thingsboard.server.service.subscription.TbSubscriptionType; |
||||
|
|
||||
|
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
import java.util.function.BiConsumer; |
||||
|
|
||||
|
@Getter |
||||
|
public abstract class AbstractNotificationSubscription<T> extends TbSubscription<T> { |
||||
|
|
||||
|
protected final AtomicInteger sequence = new AtomicInteger(); |
||||
|
protected final AtomicInteger totalUnreadCounter = new AtomicInteger(); |
||||
|
|
||||
|
public AbstractNotificationSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, TbSubscriptionType type, BiConsumer<TbSubscription<T>, T> updateProcessor) { |
||||
|
super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateProcessor); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,532 @@ |
|||||
|
/** |
||||
|
* 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.queue; |
||||
|
|
||||
|
import com.google.common.util.concurrent.ListeningExecutorService; |
||||
|
import com.google.common.util.concurrent.MoreExecutors; |
||||
|
import org.junit.jupiter.api.AfterEach; |
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.mockito.ArgumentCaptor; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.service.state.DeviceStateService; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.BDDMockito.then; |
||||
|
import static org.mockito.Mockito.doCallRealMethod; |
||||
|
import static org.mockito.Mockito.doThrow; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class DefaultTbCoreConsumerServiceTest { |
||||
|
|
||||
|
@Mock |
||||
|
private DeviceStateService stateServiceMock; |
||||
|
@Mock |
||||
|
private TbCoreConsumerStats statsMock; |
||||
|
|
||||
|
@Mock |
||||
|
private TbCallback tbCallbackMock; |
||||
|
|
||||
|
private final TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
private final DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
||||
|
private final long time = System.currentTimeMillis(); |
||||
|
|
||||
|
private ListeningExecutorService executor; |
||||
|
|
||||
|
@Mock |
||||
|
private DefaultTbCoreConsumerService defaultTbCoreConsumerServiceMock; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setup() { |
||||
|
executor = MoreExecutors.newDirectExecutorService(); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stateService", stateServiceMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "deviceActivityEventsExecutor", executor); |
||||
|
} |
||||
|
|
||||
|
@AfterEach |
||||
|
public void cleanup() { |
||||
|
if (executor != null) { |
||||
|
executor.shutdown(); |
||||
|
try { |
||||
|
if (!executor.awaitTermination(10L, TimeUnit.SECONDS)) { |
||||
|
executor.shutdownNow(); |
||||
|
} |
||||
|
} catch (InterruptedException e) { |
||||
|
executor.shutdownNow(); |
||||
|
Thread.currentThread().interrupt(); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingSuccess_whenForwardingDeviceStateMsgToStateService_thenOnSuccessCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var stateMsg = TransportProtos.DeviceStateServiceMsgProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setAdded(true) |
||||
|
.setUpdated(false) |
||||
|
.setDeleted(false) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(stateMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(stateMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(stateServiceMock).should().onQueueMsg(stateMsg, tbCallbackMock); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsEnabled_whenForwardingDeviceStateMsgToStateService_thenStatsAreRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
||||
|
|
||||
|
var stateMsg = TransportProtos.DeviceStateServiceMsgProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setAdded(true) |
||||
|
.setUpdated(false) |
||||
|
.setDeleted(false) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(stateMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(stateMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should().log(stateMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsDisabled_whenForwardingDeviceStateMsgToStateService_thenStatsAreNotRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
||||
|
|
||||
|
var stateMsg = TransportProtos.DeviceStateServiceMsgProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setAdded(true) |
||||
|
.setUpdated(false) |
||||
|
.setDeleted(false) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(stateMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(stateMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should(never()).log(stateMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingSuccess_whenForwardingConnectMsgToStateService_thenOnSuccessCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(stateServiceMock).should().onDeviceConnect(tenantId, deviceId, time); |
||||
|
then(tbCallbackMock).should().onSuccess(); |
||||
|
then(tbCallbackMock).should(never()).onFailure(any()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingFailure_whenForwardingConnectMsgToStateService_thenOnFailureCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
var runtimeException = new RuntimeException("Something bad happened!"); |
||||
|
doThrow(runtimeException).when(stateServiceMock).onDeviceConnect(tenantId, deviceId, time); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(tbCallbackMock).should(never()).onSuccess(); |
||||
|
then(tbCallbackMock).should().onFailure(runtimeException); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsEnabled_whenForwardingConnectMsgToStateService_thenStatsAreRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
||||
|
|
||||
|
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should().log(connectMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsDisabled_whenForwardingConnectMsgToStateService_thenStatsAreNotRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
||||
|
|
||||
|
var connectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(connectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should(never()).log(connectMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingSuccess_whenForwardingActivityMsgToStateService_thenOnSuccessCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(stateServiceMock).should().onDeviceActivity(tenantId, deviceId, time); |
||||
|
then(tbCallbackMock).should().onSuccess(); |
||||
|
then(tbCallbackMock).should(never()).onFailure(any()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingFailure_whenForwardingActivityMsgToStateService_thenOnFailureCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
var runtimeException = new RuntimeException("Something bad happened!"); |
||||
|
doThrow(runtimeException).when(stateServiceMock).onDeviceActivity(tenantId, deviceId, time); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(tbCallbackMock).should(never()).onSuccess(); |
||||
|
|
||||
|
var exceptionCaptor = ArgumentCaptor.forClass(Throwable.class); |
||||
|
then(tbCallbackMock).should().onFailure(exceptionCaptor.capture()); |
||||
|
assertThat(exceptionCaptor.getValue()) |
||||
|
.isInstanceOf(RuntimeException.class) |
||||
|
.hasMessage("Failed to update device activity for device [" + deviceId.getId() + "]!") |
||||
|
.hasCause(runtimeException); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsEnabled_whenForwardingActivityMsgToStateService_thenStatsAreRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
||||
|
|
||||
|
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should().log(activityMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsDisabled_whenForwardingActivityMsgToStateService_thenStatsAreNotRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
||||
|
|
||||
|
var activityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(activityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should(never()).log(activityMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingSuccess_whenForwardingDisconnectMsgToStateService_thenOnSuccessCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(stateServiceMock).should().onDeviceDisconnect(tenantId, deviceId, time); |
||||
|
then(tbCallbackMock).should().onSuccess(); |
||||
|
then(tbCallbackMock).should(never()).onFailure(any()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingFailure_whenForwardingDisconnectMsgToStateService_thenOnFailureCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
var runtimeException = new RuntimeException("Something bad happened!"); |
||||
|
doThrow(runtimeException).when(stateServiceMock).onDeviceDisconnect(tenantId, deviceId, time); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(tbCallbackMock).should(never()).onSuccess(); |
||||
|
then(tbCallbackMock).should().onFailure(runtimeException); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsEnabled_whenForwardingDisconnectMsgToStateService_thenStatsAreRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
||||
|
|
||||
|
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should().log(disconnectMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsDisabled_whenForwardingDisconnectMsgToStateService_thenStatsAreNotRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
||||
|
|
||||
|
var disconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(disconnectMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should(never()).log(disconnectMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingSuccess_whenForwardingInactivityMsgToStateService_thenOnSuccessCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(stateServiceMock).should().onDeviceInactivity(tenantId, deviceId, time); |
||||
|
then(tbCallbackMock).should().onSuccess(); |
||||
|
then(tbCallbackMock).should(never()).onFailure(any()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenProcessingFailure_whenForwardingInactivityMsgToStateService_thenOnFailureCallbackIsCalled() { |
||||
|
// GIVEN
|
||||
|
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
var runtimeException = new RuntimeException("Something bad happened!"); |
||||
|
doThrow(runtimeException).when(stateServiceMock).onDeviceInactivity(tenantId, deviceId, time); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(tbCallbackMock).should(never()).onSuccess(); |
||||
|
then(tbCallbackMock).should().onFailure(runtimeException); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsEnabled_whenForwardingInactivityMsgToStateService_thenStatsAreRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", true); |
||||
|
|
||||
|
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should().log(inactivityMsg); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void givenStatsDisabled_whenForwardingInactivityMsgToStateService_thenStatsAreNotRecorded() { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "stats", statsMock); |
||||
|
ReflectionTestUtils.setField(defaultTbCoreConsumerServiceMock, "statsEnabled", false); |
||||
|
|
||||
|
var inactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(deviceId.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(time) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(defaultTbCoreConsumerServiceMock).forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
defaultTbCoreConsumerServiceMock.forwardToStateService(inactivityMsg, tbCallbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
then(statsMock).should(never()).log(inactivityMsg); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,266 @@ |
|||||
|
/** |
||||
|
* 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.state; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.DisplayName; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.junit.jupiter.params.ParameterizedTest; |
||||
|
import org.junit.jupiter.params.provider.Arguments; |
||||
|
import org.junit.jupiter.params.provider.MethodSource; |
||||
|
import org.mockito.ArgumentCaptor; |
||||
|
import org.mockito.Captor; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.cluster.TbClusterService; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
import org.thingsboard.server.queue.TbQueueCallback; |
||||
|
import org.thingsboard.server.queue.TbQueueMsgMetadata; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
||||
|
|
||||
|
import java.util.Optional; |
||||
|
import java.util.UUID; |
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.BDDMockito.given; |
||||
|
import static org.mockito.BDDMockito.then; |
||||
|
import static org.mockito.Mockito.doThrow; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class DefaultRuleEngineDeviceStateManagerTest { |
||||
|
|
||||
|
@Mock |
||||
|
private static DeviceStateService deviceStateServiceMock; |
||||
|
@Mock |
||||
|
private static TbCallback tbCallbackMock; |
||||
|
@Mock |
||||
|
private static TbClusterService clusterServiceMock; |
||||
|
@Mock |
||||
|
private static TbQueueMsgMetadata metadataMock; |
||||
|
|
||||
|
@Mock |
||||
|
private TbServiceInfoProvider serviceInfoProviderMock; |
||||
|
@Mock |
||||
|
private PartitionService partitionServiceMock; |
||||
|
|
||||
|
@Captor |
||||
|
private static ArgumentCaptor<TbQueueCallback> queueCallbackCaptor; |
||||
|
|
||||
|
private static DefaultRuleEngineDeviceStateManager deviceStateManager; |
||||
|
|
||||
|
private static final TenantId TENANT_ID = TenantId.fromUUID(UUID.fromString("57ab2e6c-bc4c-11ee-a506-0242ac120002")); |
||||
|
private static final DeviceId DEVICE_ID = DeviceId.fromString("74a9053e-bc4c-11ee-a506-0242ac120002"); |
||||
|
private static final long EVENT_TS = System.currentTimeMillis(); |
||||
|
private static final RuntimeException RUNTIME_EXCEPTION = new RuntimeException("Something bad happened!"); |
||||
|
private static final TopicPartitionInfo MY_TPI = TopicPartitionInfo.builder().myPartition(true).build(); |
||||
|
private static final TopicPartitionInfo EXTERNAL_TPI = TopicPartitionInfo.builder().myPartition(false).build(); |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setup() { |
||||
|
deviceStateManager = new DefaultRuleEngineDeviceStateManager(serviceInfoProviderMock, partitionServiceMock, Optional.of(deviceStateServiceMock), clusterServiceMock); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@DisplayName("Given event should be routed to local service and event processed has succeeded, " + |
||||
|
"when onDeviceX() is called, then should route event to local service and call onSuccess() callback.") |
||||
|
@MethodSource |
||||
|
public void givenRoutedToLocalAndProcessingSuccess_whenOnDeviceAction_thenShouldCallLocalServiceAndSuccessCallback(Runnable onDeviceAction, Runnable actionVerification) { |
||||
|
// GIVEN
|
||||
|
given(serviceInfoProviderMock.isService(ServiceType.TB_CORE)).willReturn(true); |
||||
|
given(partitionServiceMock.resolve(ServiceType.TB_CORE, TENANT_ID, DEVICE_ID)).willReturn(MY_TPI); |
||||
|
|
||||
|
onDeviceAction.run(); |
||||
|
|
||||
|
// THEN
|
||||
|
actionVerification.run(); |
||||
|
|
||||
|
then(clusterServiceMock).shouldHaveNoInteractions(); |
||||
|
then(tbCallbackMock).should().onSuccess(); |
||||
|
then(tbCallbackMock).should(never()).onFailure(any()); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> givenRoutedToLocalAndProcessingSuccess_whenOnDeviceAction_thenShouldCallLocalServiceAndSuccessCallback() { |
||||
|
return Stream.of( |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@DisplayName("Given event should be routed to local service and event processed has failed, " + |
||||
|
"when onDeviceX() is called, then should route event to local service and call onFailure() callback.") |
||||
|
@MethodSource |
||||
|
public void givenRoutedToLocalAndProcessingFailure_whenOnDeviceAction_thenShouldCallLocalServiceAndFailureCallback( |
||||
|
Runnable exceptionThrowSetup, Runnable onDeviceAction, Runnable actionVerification |
||||
|
) { |
||||
|
// GIVEN
|
||||
|
given(serviceInfoProviderMock.isService(ServiceType.TB_CORE)).willReturn(true); |
||||
|
given(partitionServiceMock.resolve(ServiceType.TB_CORE, TENANT_ID, DEVICE_ID)).willReturn(MY_TPI); |
||||
|
|
||||
|
exceptionThrowSetup.run(); |
||||
|
|
||||
|
// WHEN
|
||||
|
onDeviceAction.run(); |
||||
|
|
||||
|
// THEN
|
||||
|
actionVerification.run(); |
||||
|
|
||||
|
then(clusterServiceMock).shouldHaveNoInteractions(); |
||||
|
then(tbCallbackMock).should(never()).onSuccess(); |
||||
|
then(tbCallbackMock).should().onFailure(RUNTIME_EXCEPTION); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> givenRoutedToLocalAndProcessingFailure_whenOnDeviceAction_thenShouldCallLocalServiceAndFailureCallback() { |
||||
|
return Stream.of( |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS), |
||||
|
(Runnable) () -> deviceStateManager.onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS), |
||||
|
(Runnable) () -> deviceStateManager.onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS), |
||||
|
(Runnable) () -> deviceStateManager.onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> doThrow(RUNTIME_EXCEPTION).when(deviceStateServiceMock).onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS), |
||||
|
(Runnable) () -> deviceStateManager.onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> then(deviceStateServiceMock).should().onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS) |
||||
|
) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@DisplayName("Given event should be routed to external service, " + |
||||
|
"when onDeviceX() is called, then should send correct queue message to external service with correct callback object.") |
||||
|
@MethodSource |
||||
|
public void givenRoutedToExternal_whenOnDeviceAction_thenShouldSendQueueMsgToExternalServiceWithCorrectCallback(Runnable onDeviceAction, Runnable actionVerification) { |
||||
|
// WHEN
|
||||
|
ReflectionTestUtils.setField(deviceStateManager, "deviceStateService", Optional.empty()); |
||||
|
given(serviceInfoProviderMock.isService(ServiceType.TB_CORE)).willReturn(false); |
||||
|
given(partitionServiceMock.resolve(ServiceType.TB_CORE, TENANT_ID, DEVICE_ID)).willReturn(EXTERNAL_TPI); |
||||
|
|
||||
|
onDeviceAction.run(); |
||||
|
|
||||
|
// THEN
|
||||
|
actionVerification.run(); |
||||
|
|
||||
|
TbQueueCallback callback = queueCallbackCaptor.getValue(); |
||||
|
callback.onSuccess(metadataMock); |
||||
|
then(tbCallbackMock).should().onSuccess(); |
||||
|
callback.onFailure(RUNTIME_EXCEPTION); |
||||
|
then(tbCallbackMock).should().onFailure(RUNTIME_EXCEPTION); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> givenRoutedToExternal_whenOnDeviceAction_thenShouldSendQueueMsgToExternalServiceWithCorrectCallback() { |
||||
|
return Stream.of( |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceConnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> { |
||||
|
var deviceConnectMsg = TransportProtos.DeviceConnectProto.newBuilder() |
||||
|
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
||||
|
.setLastConnectTime(EVENT_TS) |
||||
|
.build(); |
||||
|
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceConnectMsg(deviceConnectMsg) |
||||
|
.build(); |
||||
|
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
||||
|
} |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceActivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> { |
||||
|
var deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder() |
||||
|
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
||||
|
.setLastActivityTime(EVENT_TS) |
||||
|
.build(); |
||||
|
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceActivityMsg(deviceActivityMsg) |
||||
|
.build(); |
||||
|
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
||||
|
} |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceDisconnect(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> { |
||||
|
var deviceDisconnectMsg = TransportProtos.DeviceDisconnectProto.newBuilder() |
||||
|
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
||||
|
.setLastDisconnectTime(EVENT_TS) |
||||
|
.build(); |
||||
|
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceDisconnectMsg(deviceDisconnectMsg) |
||||
|
.build(); |
||||
|
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
||||
|
} |
||||
|
), |
||||
|
Arguments.of( |
||||
|
(Runnable) () -> deviceStateManager.onDeviceInactivity(TENANT_ID, DEVICE_ID, EVENT_TS, tbCallbackMock), |
||||
|
(Runnable) () -> { |
||||
|
var deviceInactivityMsg = TransportProtos.DeviceInactivityProto.newBuilder() |
||||
|
.setTenantIdMSB(TENANT_ID.getId().getMostSignificantBits()) |
||||
|
.setTenantIdLSB(TENANT_ID.getId().getLeastSignificantBits()) |
||||
|
.setDeviceIdMSB(DEVICE_ID.getId().getMostSignificantBits()) |
||||
|
.setDeviceIdLSB(DEVICE_ID.getId().getLeastSignificantBits()) |
||||
|
.setLastInactivityTime(EVENT_TS) |
||||
|
.build(); |
||||
|
var toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
||||
|
.setDeviceInactivityMsg(deviceInactivityMsg) |
||||
|
.build(); |
||||
|
then(clusterServiceMock).should().pushMsgToCore(eq(EXTERNAL_TPI), any(UUID.class), eq(toCoreMsg), queueCallbackCaptor.capture()); |
||||
|
} |
||||
|
) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,23 @@ |
|||||
|
/** |
||||
|
* 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.common.data.mobile; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
@Data |
||||
|
public class MobileSessionInfo { |
||||
|
private long fcmTokenTimestamp; |
||||
|
} |
||||
@ -0,0 +1,27 @@ |
|||||
|
/** |
||||
|
* 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.common.data.mobile; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.util.Map; |
||||
|
|
||||
|
@Data |
||||
|
public class UserMobileInfo { |
||||
|
|
||||
|
private Map<String, MobileSessionInfo> sessions; |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,35 @@ |
|||||
|
/** |
||||
|
* 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.common.data.notification.settings; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
||||
|
|
||||
|
import javax.validation.constraints.NotEmpty; |
||||
|
|
||||
|
@Data |
||||
|
public class MobileAppNotificationDeliveryMethodConfig implements NotificationDeliveryMethodConfig { |
||||
|
|
||||
|
private String firebaseServiceAccountCredentialsFileName; |
||||
|
@NotEmpty |
||||
|
private String firebaseServiceAccountCredentials; |
||||
|
|
||||
|
@Override |
||||
|
public NotificationDeliveryMethod getMethod() { |
||||
|
return NotificationDeliveryMethod.MOBILE_APP; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,64 @@ |
|||||
|
/** |
||||
|
* 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.common.data.notification.template; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import lombok.Data; |
||||
|
import lombok.EqualsAndHashCode; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import lombok.ToString; |
||||
|
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
||||
|
|
||||
|
import javax.validation.constraints.NotEmpty; |
||||
|
import java.util.List; |
||||
|
|
||||
|
@Data |
||||
|
@NoArgsConstructor |
||||
|
@EqualsAndHashCode(callSuper = true) |
||||
|
@ToString(callSuper = true) |
||||
|
public class MobileAppDeliveryMethodNotificationTemplate extends DeliveryMethodNotificationTemplate implements HasSubject { |
||||
|
|
||||
|
@NotEmpty |
||||
|
private String subject; |
||||
|
private JsonNode additionalConfig; |
||||
|
|
||||
|
private final List<TemplatableValue> templatableValues = List.of( |
||||
|
TemplatableValue.of(this::getBody, this::setBody), |
||||
|
TemplatableValue.of(this::getSubject, this::setSubject) |
||||
|
); |
||||
|
|
||||
|
public MobileAppDeliveryMethodNotificationTemplate(MobileAppDeliveryMethodNotificationTemplate other) { |
||||
|
super(other); |
||||
|
this.subject = other.subject; |
||||
|
this.additionalConfig = other.additionalConfig; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public NotificationDeliveryMethod getMethod() { |
||||
|
return NotificationDeliveryMethod.MOBILE_APP; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public MobileAppDeliveryMethodNotificationTemplate copy() { |
||||
|
return new MobileAppDeliveryMethodNotificationTemplate(this); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public List<TemplatableValue> getTemplatableValues() { |
||||
|
return templatableValues; |
||||
|
} |
||||
|
|
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue