Vladyslav_Prykhodko
3 years ago
311 changed files with 10073 additions and 2020 deletions
File diff suppressed because one or more lines are too long
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,29 @@ |
|||||
|
/** |
||||
|
* 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.edge.rpc.constructor.alarm; |
||||
|
|
||||
|
import org.thingsboard.common.util.JacksonUtil; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmComment; |
||||
|
import org.thingsboard.server.gen.edge.v1.AlarmCommentUpdateMsg; |
||||
|
import org.thingsboard.server.gen.edge.v1.UpdateMsgType; |
||||
|
|
||||
|
public abstract class BaseAlarmMsgConstructor implements AlarmMsgConstructor { |
||||
|
|
||||
|
@Override |
||||
|
public AlarmCommentUpdateMsg constructAlarmCommentUpdatedMsg(UpdateMsgType msgType, AlarmComment alarmComment) { |
||||
|
return AlarmCommentUpdateMsg.newBuilder().setMsgType(msgType).setEntity(JacksonUtil.toString(alarmComment)).build(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,74 @@ |
|||||
|
/** |
||||
|
* 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.actors.tenant; |
||||
|
|
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.thingsboard.server.actors.ActorSystemContext; |
||||
|
import org.thingsboard.server.actors.TbActorCtx; |
||||
|
import org.thingsboard.server.actors.TbActorRef; |
||||
|
import org.thingsboard.server.common.data.id.DeviceId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
||||
|
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg; |
||||
|
import org.thingsboard.server.dao.tenant.TenantService; |
||||
|
|
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.reset; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
|
||||
|
public class TenantActorTest { |
||||
|
|
||||
|
TenantActor tenantActor; |
||||
|
TbActorCtx ctx; |
||||
|
ActorSystemContext systemContext; |
||||
|
TenantId tenantId = TenantId.SYS_TENANT_ID; |
||||
|
DeviceId deviceId = DeviceId.fromString("78bf9b26-74ef-4af2-9cfb-ad6cf24ad2ec"); |
||||
|
|
||||
|
@Before |
||||
|
public void setUp() throws Exception { |
||||
|
systemContext = mock(ActorSystemContext.class); |
||||
|
ctx = mock(TbActorCtx.class); |
||||
|
tenantActor = (TenantActor) new TenantActor.ActorCreator(systemContext, tenantId).createActor(); |
||||
|
when(systemContext.getTenantService()).thenReturn(mock(TenantService.class)); |
||||
|
tenantActor.init(ctx); |
||||
|
tenantActor.cantFindTenant = false; |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void deleteDeviceTest() { |
||||
|
TbActorRef deviceActorRef = mock(TbActorRef.class); |
||||
|
when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 0,true)); |
||||
|
when(ctx.getOrCreateChildActor(any(), any(), any(), any())).thenReturn(deviceActorRef); |
||||
|
ComponentLifecycleMsg componentLifecycleMsg = new ComponentLifecycleMsg(tenantId, deviceId, ComponentLifecycleEvent.DELETED); |
||||
|
tenantActor.doProcess(componentLifecycleMsg); |
||||
|
verify(deviceActorRef).tellWithHighPriority(eq(new DeviceDeleteMsg(tenantId, deviceId))); |
||||
|
|
||||
|
reset(ctx, deviceActorRef); |
||||
|
when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 1,false)); |
||||
|
tenantActor.doProcess(componentLifecycleMsg); |
||||
|
verify(ctx, never()).getOrCreateChildActor(any(), any(), any(), any()); |
||||
|
verify(deviceActorRef, never()).tellWithHighPriority(any()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,92 @@ |
|||||
|
/** |
||||
|
* 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.kv; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.EqualsAndHashCode; |
||||
|
import lombok.Getter; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.thingsboard.server.common.data.StringUtils; |
||||
|
|
||||
|
import java.time.DateTimeException; |
||||
|
import java.time.ZoneId; |
||||
|
import java.util.Map; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@AllArgsConstructor |
||||
|
@EqualsAndHashCode |
||||
|
@Slf4j |
||||
|
public class AggregationParams { |
||||
|
private static final Map<String, String> TZ_LINKS = Map.of("EST", "America/New_York", "GMT+0", "GMT", "GMT-0", "GMT", "HST", "US/Hawaii", "MST", "America/Phoenix", "ROC", "Asia/Taipei"); |
||||
|
@Getter |
||||
|
private final Aggregation aggregation; |
||||
|
@Getter |
||||
|
private final IntervalType intervalType; |
||||
|
@Getter |
||||
|
private final ZoneId tzId; |
||||
|
|
||||
|
private final long interval; |
||||
|
|
||||
|
public static AggregationParams none() { |
||||
|
return new AggregationParams(Aggregation.NONE, null, null, 0L); |
||||
|
} |
||||
|
|
||||
|
public static AggregationParams milliseconds(Aggregation aggregationType, long aggregationIntervalMs) { |
||||
|
return new AggregationParams(aggregationType, IntervalType.MILLISECONDS, null, aggregationIntervalMs); |
||||
|
} |
||||
|
|
||||
|
public static AggregationParams calendar(Aggregation aggregationType, IntervalType intervalType, String tzIdStr) { |
||||
|
return calendar(aggregationType, intervalType, getZoneId(tzIdStr)); |
||||
|
} |
||||
|
|
||||
|
public static AggregationParams calendar(Aggregation aggregationType, IntervalType intervalType, ZoneId tzId) { |
||||
|
return new AggregationParams(aggregationType, intervalType, tzId, 0L); |
||||
|
} |
||||
|
|
||||
|
public static AggregationParams of(Aggregation aggregation, IntervalType intervalType, ZoneId tzId, long interval) { |
||||
|
return new AggregationParams(aggregation, intervalType, tzId, interval); |
||||
|
} |
||||
|
|
||||
|
public long getInterval() { |
||||
|
if (intervalType == null) { |
||||
|
return 0L; |
||||
|
} else { |
||||
|
switch (intervalType) { |
||||
|
case WEEK: |
||||
|
case WEEK_ISO: |
||||
|
return TimeUnit.DAYS.toMillis(7); |
||||
|
case MONTH: |
||||
|
return TimeUnit.DAYS.toMillis(30); |
||||
|
case QUARTER: |
||||
|
return TimeUnit.DAYS.toMillis(90); |
||||
|
default: |
||||
|
return interval; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private static ZoneId getZoneId(String tzIdStr) { |
||||
|
if (StringUtils.isEmpty(tzIdStr)) { |
||||
|
return ZoneId.systemDefault(); |
||||
|
} |
||||
|
try { |
||||
|
return ZoneId.of(tzIdStr, TZ_LINKS); |
||||
|
} catch (DateTimeException e) { |
||||
|
log.warn("[{}] Failed to convert the time zone. Fallback to default.", tzIdStr); |
||||
|
return ZoneId.systemDefault(); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,22 @@ |
|||||
|
/** |
||||
|
* 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.kv; |
||||
|
|
||||
|
public enum IntervalType { |
||||
|
|
||||
|
MILLISECONDS, WEEK/*Sunday-Saturday*/, WEEK_ISO/*Monday-Sunday*/, MONTH, QUARTER |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,176 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy; |
||||
|
import org.thingsboard.server.queue.scheduler.SchedulerComponent; |
||||
|
|
||||
|
import java.util.Map; |
||||
|
import java.util.Random; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.atomic.AtomicBoolean; |
||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class AbstractActivityManager<Key, Metadata> implements ActivityManager<Key, Metadata> { |
||||
|
|
||||
|
private final ConcurrentMap<Key, ActivityStateWrapper> states = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
@Autowired |
||||
|
protected SchedulerComponent scheduler; |
||||
|
|
||||
|
@Data |
||||
|
private class ActivityStateWrapper { |
||||
|
|
||||
|
private volatile ActivityState<Metadata> state; |
||||
|
private volatile long lastReportedTime; |
||||
|
private volatile ActivityStrategy strategy; |
||||
|
|
||||
|
} |
||||
|
|
||||
|
protected void init() { |
||||
|
var reportingPeriodMillis = getReportingPeriodMillis(); |
||||
|
scheduler.scheduleAtFixedRate(this::onReportingPeriodEnd, new Random().nextInt((int) reportingPeriodMillis), reportingPeriodMillis, TimeUnit.MILLISECONDS); |
||||
|
} |
||||
|
|
||||
|
protected abstract long getReportingPeriodMillis(); |
||||
|
|
||||
|
protected abstract ActivityStrategy getStrategy(); |
||||
|
|
||||
|
protected abstract ActivityState<Metadata> updateState(Key key, ActivityState<Metadata> state); |
||||
|
|
||||
|
protected abstract boolean hasExpired(long lastRecordedTime); |
||||
|
|
||||
|
protected abstract void onStateExpiry(Key key, Metadata metadata); |
||||
|
|
||||
|
protected abstract void reportActivity(Key key, Metadata metadata, long timeToReport, ActivityReportCallback<Key> callback); |
||||
|
|
||||
|
@Override |
||||
|
public void onActivity(Key key, Metadata metadata, long newLastRecordedTime) { |
||||
|
if (key == null) { |
||||
|
log.error("Failed to process activity event: provided activity key is null."); |
||||
|
return; |
||||
|
} |
||||
|
log.debug("Received activity event for key: [{}]", key); |
||||
|
|
||||
|
var shouldReport = new AtomicBoolean(false); |
||||
|
var lastRecordedTime = new AtomicLong(); |
||||
|
var lastReportedTime = new AtomicLong(); |
||||
|
|
||||
|
states.compute(key, (__, stateWrapper) -> { |
||||
|
if (stateWrapper == null) { |
||||
|
ActivityState<Metadata> newState = new ActivityState<>(); |
||||
|
stateWrapper = new ActivityStateWrapper(); |
||||
|
stateWrapper.setState(newState); |
||||
|
stateWrapper.setStrategy(getStrategy()); |
||||
|
} |
||||
|
var state = stateWrapper.getState(); |
||||
|
state.setMetadata(metadata); |
||||
|
if (state.getLastRecordedTime() < newLastRecordedTime) { |
||||
|
state.setLastRecordedTime(newLastRecordedTime); |
||||
|
} |
||||
|
shouldReport.set(stateWrapper.getStrategy().onActivity()); |
||||
|
lastRecordedTime.set(state.getLastRecordedTime()); |
||||
|
lastReportedTime.set(stateWrapper.getLastReportedTime()); |
||||
|
return stateWrapper; |
||||
|
}); |
||||
|
|
||||
|
if (shouldReport.get() && lastReportedTime.get() < lastRecordedTime.get()) { |
||||
|
log.debug("Going to report first activity event for key: [{}].", key); |
||||
|
reportActivity(key, metadata, lastRecordedTime.get(), new ActivityReportCallback<>() { |
||||
|
@Override |
||||
|
public void onSuccess(Key key, long reportedTime) { |
||||
|
updateLastReportedTime(key, reportedTime); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Key key, Throwable t) { |
||||
|
log.debug("Failed to report first activity event for key: [{}].", key, t); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onReportingPeriodEnd() { |
||||
|
log.debug("Going to end reporting period."); |
||||
|
for (Map.Entry<Key, ActivityStateWrapper> entry : states.entrySet()) { |
||||
|
var key = entry.getKey(); |
||||
|
var stateWrapper = entry.getValue(); |
||||
|
var currentState = stateWrapper.getState(); |
||||
|
|
||||
|
long lastRecordedTime = currentState.getLastRecordedTime(); |
||||
|
long lastReportedTime = stateWrapper.getLastReportedTime(); |
||||
|
var metadata = currentState.getMetadata(); |
||||
|
|
||||
|
boolean hasExpired; |
||||
|
boolean shouldReport; |
||||
|
|
||||
|
var updatedState = updateState(key, currentState); |
||||
|
if (updatedState != null) { |
||||
|
stateWrapper.setState(updatedState); |
||||
|
lastRecordedTime = updatedState.getLastRecordedTime(); |
||||
|
metadata = updatedState.getMetadata(); |
||||
|
hasExpired = hasExpired(lastRecordedTime); |
||||
|
shouldReport = stateWrapper.getStrategy().onReportingPeriodEnd(); |
||||
|
} else { |
||||
|
states.remove(key); |
||||
|
hasExpired = false; |
||||
|
shouldReport = true; |
||||
|
} |
||||
|
|
||||
|
if (hasExpired) { |
||||
|
states.remove(key); |
||||
|
onStateExpiry(key, metadata); |
||||
|
shouldReport = true; |
||||
|
} |
||||
|
|
||||
|
if (shouldReport && lastReportedTime < lastRecordedTime) { |
||||
|
log.debug("Going to report last activity event for key: [{}].", key); |
||||
|
reportActivity(key, metadata, lastRecordedTime, new ActivityReportCallback<>() { |
||||
|
@Override |
||||
|
public void onSuccess(Key key, long reportedTime) { |
||||
|
updateLastReportedTime(key, reportedTime); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Key key, Throwable t) { |
||||
|
log.debug("Failed to report last activity event for key: [{}].", key, t); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public long getLastRecordedTime(Key key) { |
||||
|
ActivityStateWrapper stateWrapper = states.get(key); |
||||
|
return stateWrapper == null ? 0L : stateWrapper.getState().getLastRecordedTime(); |
||||
|
} |
||||
|
|
||||
|
private void updateLastReportedTime(Key key, long newLastReportedTime) { |
||||
|
states.computeIfPresent(key, (__, stateWrapper) -> { |
||||
|
stateWrapper.setLastReportedTime(Math.max(stateWrapper.getLastReportedTime(), newLastReportedTime)); |
||||
|
return stateWrapper; |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity; |
||||
|
|
||||
|
public interface ActivityManager<Key, Metadata> { |
||||
|
|
||||
|
void onActivity(Key key, Metadata metadata, long activityTimeMillis); |
||||
|
|
||||
|
void onReportingPeriodEnd(); |
||||
|
|
||||
|
long getLastRecordedTime(Key key); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,24 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity; |
||||
|
|
||||
|
public interface ActivityReportCallback<Key> { |
||||
|
|
||||
|
void onSuccess(Key key, long reportedTime); |
||||
|
|
||||
|
void onFailure(Key key, Throwable t); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,24 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
public interface ActivityStrategy { |
||||
|
|
||||
|
boolean onActivity(); |
||||
|
|
||||
|
boolean onReportingPeriodEnd(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,47 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
public enum ActivityStrategyType { |
||||
|
|
||||
|
ALL { |
||||
|
@Override |
||||
|
public ActivityStrategy toStrategy() { |
||||
|
return AllEventsActivityStrategy.getInstance(); |
||||
|
} |
||||
|
}, |
||||
|
FIRST { |
||||
|
@Override |
||||
|
public ActivityStrategy toStrategy() { |
||||
|
return new FirstEventActivityStrategy(); |
||||
|
} |
||||
|
}, |
||||
|
LAST { |
||||
|
@Override |
||||
|
public ActivityStrategy toStrategy() { |
||||
|
return LastEventActivityStrategy.getInstance(); |
||||
|
} |
||||
|
}, |
||||
|
FIRST_AND_LAST { |
||||
|
@Override |
||||
|
public ActivityStrategy toStrategy() { |
||||
|
return new FirstAndLastEventActivityStrategy(); |
||||
|
} |
||||
|
}; |
||||
|
|
||||
|
public abstract ActivityStrategy toStrategy(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,40 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import lombok.EqualsAndHashCode; |
||||
|
|
||||
|
@EqualsAndHashCode |
||||
|
public final class FirstAndLastEventActivityStrategy implements ActivityStrategy { |
||||
|
|
||||
|
private boolean firstEventReceived; |
||||
|
|
||||
|
@Override |
||||
|
public synchronized boolean onActivity() { |
||||
|
if (!firstEventReceived) { |
||||
|
firstEventReceived = true; |
||||
|
return true; |
||||
|
} |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public synchronized boolean onReportingPeriodEnd() { |
||||
|
firstEventReceived = false; |
||||
|
return true; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,40 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import lombok.EqualsAndHashCode; |
||||
|
|
||||
|
@EqualsAndHashCode |
||||
|
public final class FirstEventActivityStrategy implements ActivityStrategy { |
||||
|
|
||||
|
private boolean firstEventReceived; |
||||
|
|
||||
|
@Override |
||||
|
public synchronized boolean onActivity() { |
||||
|
if (!firstEventReceived) { |
||||
|
firstEventReceived = true; |
||||
|
return true; |
||||
|
} |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public synchronized boolean onReportingPeriodEnd() { |
||||
|
firstEventReceived = false; |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,39 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
public final class LastEventActivityStrategy implements ActivityStrategy { |
||||
|
|
||||
|
private static final LastEventActivityStrategy INSTANCE = new LastEventActivityStrategy(); |
||||
|
|
||||
|
private LastEventActivityStrategy() { |
||||
|
} |
||||
|
|
||||
|
public static LastEventActivityStrategy getInstance() { |
||||
|
return INSTANCE; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean onActivity() { |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public boolean onReportingPeriodEnd() { |
||||
|
return true; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,138 @@ |
|||||
|
/** |
||||
|
* 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.transport.service; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.thingsboard.server.common.transport.TransportService; |
||||
|
import org.thingsboard.server.common.transport.TransportServiceCallback; |
||||
|
import org.thingsboard.server.common.transport.activity.AbstractActivityManager; |
||||
|
import org.thingsboard.server.common.transport.activity.ActivityReportCallback; |
||||
|
import org.thingsboard.server.common.transport.activity.ActivityState; |
||||
|
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy; |
||||
|
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategyType; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class TransportActivityManager extends AbstractActivityManager<UUID, TransportProtos.SessionInfoProto> implements TransportService { |
||||
|
|
||||
|
public static final String SESSION_EXPIRED_MESSAGE = "Session has expired due to last activity time!"; |
||||
|
|
||||
|
public static final TransportProtos.SessionEventMsg SESSION_EVENT_MSG_CLOSED = TransportProtos.SessionEventMsg.newBuilder() |
||||
|
.setSessionType(TransportProtos.SessionType.ASYNC) |
||||
|
.setEvent(TransportProtos.SessionEvent.CLOSED).build(); |
||||
|
public static final TransportProtos.SessionCloseNotificationProto SESSION_EXPIRED_NOTIFICATION_PROTO = TransportProtos.SessionCloseNotificationProto.newBuilder() |
||||
|
.setMessage(SESSION_EXPIRED_MESSAGE).build(); |
||||
|
|
||||
|
public final ConcurrentMap<UUID, SessionMetaData> sessions = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
@Value("${transport.sessions.report_timeout}") |
||||
|
protected long sessionReportTimeout; |
||||
|
|
||||
|
@Value("${transport.sessions.inactivity_timeout}") |
||||
|
protected long sessionInactivityTimeout; |
||||
|
|
||||
|
@Value("${transport.activity.reporting_strategy:LAST}") |
||||
|
private ActivityStrategyType reportingStrategyType; |
||||
|
|
||||
|
@Override |
||||
|
protected long getReportingPeriodMillis() { |
||||
|
return sessionReportTimeout; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected ActivityStrategy getStrategy() { |
||||
|
return reportingStrategyType.toStrategy(); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected ActivityState<TransportProtos.SessionInfoProto> updateState(UUID sessionId, ActivityState<TransportProtos.SessionInfoProto> state) { |
||||
|
SessionMetaData session = sessions.get(sessionId); |
||||
|
if (session == null) { |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
state.setMetadata(session.getSessionInfo()); |
||||
|
var sessionInfo = state.getMetadata(); |
||||
|
|
||||
|
if (sessionInfo.getGwSessionIdMSB() == 0L || sessionInfo.getGwSessionIdLSB() == 0L) { |
||||
|
return state; |
||||
|
} |
||||
|
|
||||
|
var gwSessionId = new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB()); |
||||
|
SessionMetaData gwSession = sessions.get(gwSessionId); |
||||
|
if (gwSession == null || !gwSession.isOverwriteActivityTime()) { |
||||
|
return state; |
||||
|
} |
||||
|
|
||||
|
long lastRecordedTime = state.getLastRecordedTime(); |
||||
|
long gwLastRecordedTime = getLastRecordedTime(gwSessionId); |
||||
|
log.debug("Session with id: [{}] has gateway session with id: [{}] with overwrite activity time enabled. " + |
||||
|
"Updating last activity time. Session last recorded time: [{}], gateway session last recorded time: [{}].", |
||||
|
sessionId, gwSessionId, lastRecordedTime, gwLastRecordedTime); |
||||
|
state.setLastRecordedTime(Math.max(lastRecordedTime, gwLastRecordedTime)); |
||||
|
return state; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected boolean hasExpired(long lastRecordedTime) { |
||||
|
return (getCurrentTimeMillis() - sessionInactivityTimeout) > lastRecordedTime; |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected void onStateExpiry(UUID sessionId, TransportProtos.SessionInfoProto sessionInfo) { |
||||
|
log.debug("Session with id: [{}] has expired due to last activity time.", sessionId); |
||||
|
SessionMetaData expiredSession = sessions.remove(sessionId); |
||||
|
if (expiredSession != null) { |
||||
|
deregisterSession(sessionInfo); |
||||
|
process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); |
||||
|
expiredSession.getListener().onRemoteSessionCloseCommand(sessionId, SESSION_EXPIRED_NOTIFICATION_PROTO); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected void reportActivity(UUID sessionId, TransportProtos.SessionInfoProto currentSessionInfo, long timeToReport, ActivityReportCallback<UUID> callback) { |
||||
|
log.debug("Reporting activity state for session with id: [{}]. Time to report: [{}].", sessionId, timeToReport); |
||||
|
SessionMetaData session = sessions.get(sessionId); |
||||
|
TransportProtos.SubscriptionInfoProto subscriptionInfo = TransportProtos.SubscriptionInfoProto.newBuilder() |
||||
|
.setAttributeSubscription(session != null && session.isSubscribedToAttributes()) |
||||
|
.setRpcSubscription(session != null && session.isSubscribedToRPC()) |
||||
|
.setLastActivityTime(timeToReport) |
||||
|
.build(); |
||||
|
TransportProtos.SessionInfoProto sessionInfo = session != null ? session.getSessionInfo() : currentSessionInfo; |
||||
|
process(sessionInfo, subscriptionInfo, new TransportServiceCallback<>() { |
||||
|
@Override |
||||
|
public void onSuccess(Void msgAcknowledged) { |
||||
|
callback.onSuccess(sessionId, timeToReport); |
||||
|
|
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onError(Throwable e) { |
||||
|
callback.onFailure(sessionId, e); |
||||
|
} |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
protected long getCurrentTimeMillis() { |
||||
|
return System.currentTimeMillis(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,44 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.api.Test; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
|
||||
|
public class ActivityStrategyTypeTest { |
||||
|
|
||||
|
@Test |
||||
|
public void testCreateAllEventsStrategy() { |
||||
|
assertThat(ActivityStrategyType.ALL.toStrategy()).isEqualTo(AllEventsActivityStrategy.getInstance()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testCreateFirstEventStrategy() { |
||||
|
assertThat(ActivityStrategyType.FIRST.toStrategy()).isEqualTo(new FirstEventActivityStrategy()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testCreateLastEventStrategy() { |
||||
|
assertThat(ActivityStrategyType.LAST.toStrategy()).isEqualTo(LastEventActivityStrategy.getInstance()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testCreateFirstAndLastEventStrategy() { |
||||
|
assertThat(ActivityStrategyType.FIRST_AND_LAST.toStrategy()).isEqualTo(new FirstAndLastEventActivityStrategy()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,42 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
|
||||
|
import static org.junit.jupiter.api.Assertions.assertTrue; |
||||
|
|
||||
|
public class AllEventsActivityStrategyTest { |
||||
|
|
||||
|
private AllEventsActivityStrategy strategy; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
strategy = AllEventsActivityStrategy.getInstance(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnActivity() { |
||||
|
assertTrue(strategy.onActivity(), "onActivity() should always return true."); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnReportingPeriodEnd() { |
||||
|
assertTrue(strategy.onReportingPeriodEnd(), "onReportingPeriodEnd() should always return true."); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,53 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
|
||||
|
import static org.junit.jupiter.api.Assertions.assertFalse; |
||||
|
import static org.junit.jupiter.api.Assertions.assertTrue; |
||||
|
|
||||
|
public class FirstAndLastEventActivityStrategyTest { |
||||
|
|
||||
|
private FirstAndLastEventActivityStrategy strategy; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
strategy = new FirstAndLastEventActivityStrategy(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnActivity_FirstCall() { |
||||
|
assertTrue(strategy.onActivity(), "First call of onActivity() should return true."); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnActivity_SubsequentCalls() { |
||||
|
assertTrue(strategy.onActivity(), "First call of onActivity() should return true."); |
||||
|
assertFalse(strategy.onActivity(), "Subsequent calls of onActivity() should return false."); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnReportingPeriodEnd() { |
||||
|
assertTrue(strategy.onActivity(), "First call of onActivity() should return true."); |
||||
|
assertTrue(strategy.onReportingPeriodEnd(), "onReportingPeriodEnd() should always return true."); |
||||
|
assertTrue(strategy.onActivity(), "onActivity() should return true after onReportingPeriodEnd() for the next reporting period"); |
||||
|
assertTrue(strategy.onReportingPeriodEnd(), "onReportingPeriodEnd() should always return true."); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
} |
||||
@ -0,0 +1,52 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
|
||||
|
import static org.junit.jupiter.api.Assertions.assertFalse; |
||||
|
import static org.junit.jupiter.api.Assertions.assertTrue; |
||||
|
|
||||
|
public class FirstEventActivityStrategyTest { |
||||
|
|
||||
|
private FirstEventActivityStrategy strategy; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
strategy = new FirstEventActivityStrategy(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnActivity_FirstCall() { |
||||
|
assertTrue(strategy.onActivity(), "First call of onActivity() should return true."); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnActivity_SubsequentCalls() { |
||||
|
assertTrue(strategy.onActivity(), "First call of onActivity() should return true."); |
||||
|
assertFalse(strategy.onActivity(), "Subsequent calls of onActivity() should return false."); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnReportingPeriodEnd() { |
||||
|
assertTrue(strategy.onActivity(), "First call of onActivity() should return true."); |
||||
|
assertFalse(strategy.onReportingPeriodEnd(), "onReportingPeriodEnd() should always return false."); |
||||
|
assertTrue(strategy.onActivity(), "onActivity() should return true after onReportingPeriodEnd()."); |
||||
|
assertFalse(strategy.onReportingPeriodEnd(), "onReportingPeriodEnd() should always return false."); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,43 @@ |
|||||
|
/** |
||||
|
* 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.transport.activity.strategy; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
|
||||
|
import static org.junit.jupiter.api.Assertions.assertFalse; |
||||
|
import static org.junit.jupiter.api.Assertions.assertTrue; |
||||
|
|
||||
|
public class LastEventActivityStrategyTest { |
||||
|
|
||||
|
private LastEventActivityStrategy strategy; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setUp() { |
||||
|
strategy = LastEventActivityStrategy.getInstance(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnActivity() { |
||||
|
assertFalse(strategy.onActivity(), "onActivity() should always return false."); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testOnReportingPeriodEnd() { |
||||
|
assertTrue(strategy.onReportingPeriodEnd(), "onReportingPeriodEnd() should always return true."); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,517 @@ |
|||||
|
/** |
||||
|
* 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.transport.service; |
||||
|
|
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
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.EnumSource; |
||||
|
import org.junit.jupiter.params.provider.MethodSource; |
||||
|
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.transport.SessionMsgListener; |
||||
|
import org.thingsboard.server.common.transport.TransportServiceCallback; |
||||
|
import org.thingsboard.server.common.transport.activity.ActivityReportCallback; |
||||
|
import org.thingsboard.server.common.transport.activity.ActivityState; |
||||
|
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategy; |
||||
|
import org.thingsboard.server.common.transport.activity.strategy.ActivityStrategyType; |
||||
|
import org.thingsboard.server.gen.transport.TransportProtos; |
||||
|
|
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.ConcurrentMap; |
||||
|
import java.util.stream.Stream; |
||||
|
|
||||
|
import static org.assertj.core.api.Assertions.assertThat; |
||||
|
import static org.mockito.ArgumentMatchers.any; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.Mockito.doCallRealMethod; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.never; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED; |
||||
|
import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EXPIRED_NOTIFICATION_PROTO; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class TransportActivityManagerTest { |
||||
|
|
||||
|
private final UUID SESSION_ID = UUID.fromString("1306648a-9b26-11ee-b9d1-0242ac120002"); |
||||
|
|
||||
|
@Mock |
||||
|
private DefaultTransportService transportServiceMock; |
||||
|
private ConcurrentMap<UUID, SessionMetaData> sessions; |
||||
|
|
||||
|
@BeforeEach |
||||
|
public void setup() { |
||||
|
sessions = new ConcurrentHashMap<>(); |
||||
|
ReflectionTestUtils.setField(transportServiceMock, "sessions", sessions); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenFirstActivityForAlreadyRemovedSessionAndFirstEventReportingStrategy_whenOnActivity_thenShouldRecordActivityAndReport() { |
||||
|
// GIVEN
|
||||
|
ConcurrentMap<UUID, Object> states = new ConcurrentHashMap<>(); |
||||
|
ReflectionTestUtils.setField(transportServiceMock, "states", states); |
||||
|
|
||||
|
var strategyMock = mock(ActivityStrategy.class); |
||||
|
when(transportServiceMock.getStrategy()).thenReturn(strategyMock); |
||||
|
when(strategyMock.onActivity()).thenReturn(true); |
||||
|
|
||||
|
long activityTime = 123L; |
||||
|
var sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(transportServiceMock).getLastRecordedTime(SESSION_ID); |
||||
|
doCallRealMethod().when(transportServiceMock).onActivity(SESSION_ID, sessionInfo, activityTime); |
||||
|
|
||||
|
// WHEN
|
||||
|
transportServiceMock.onActivity(SESSION_ID, sessionInfo, activityTime); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(states).containsKey(SESSION_ID); |
||||
|
assertThat(transportServiceMock.getLastRecordedTime(SESSION_ID)).isEqualTo(activityTime); |
||||
|
verify(transportServiceMock).reportActivity(eq(SESSION_ID), eq(sessionInfo), eq(activityTime), any(ActivityReportCallback.class)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenKeyAndTimeToReportAndSessionExists_whenReportingActivity_thenShouldReportActivityWithSubscriptionsAndSessionInfoFromSession() { |
||||
|
// GIVEN
|
||||
|
long expectedTime = 123L; |
||||
|
boolean expectedAttributesSubscription = true; |
||||
|
boolean expectedRPCSubscription = true; |
||||
|
TransportProtos.SessionInfoProto expectedSessionInfo = TransportProtos.SessionInfoProto.getDefaultInstance(); |
||||
|
|
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
SessionMetaData session = new SessionMetaData(expectedSessionInfo, TransportProtos.SessionType.ASYNC, listenerMock); |
||||
|
session.setSubscribedToAttributes(expectedAttributesSubscription); |
||||
|
session.setSubscribedToRPC(expectedRPCSubscription); |
||||
|
sessions.put(SESSION_ID, session); |
||||
|
|
||||
|
ActivityReportCallback<UUID> callbackMock = mock(ActivityReportCallback.class); |
||||
|
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
|
||||
|
doCallRealMethod().when(transportServiceMock).reportActivity(SESSION_ID, sessionInfo, expectedTime, callbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
transportServiceMock.reportActivity(SESSION_ID, sessionInfo, expectedTime, callbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
ArgumentCaptor<TransportProtos.SessionInfoProto> sessionInfoCaptor = ArgumentCaptor.forClass(TransportProtos.SessionInfoProto.class); |
||||
|
ArgumentCaptor<TransportProtos.SubscriptionInfoProto> subscriptionInfoCaptor = ArgumentCaptor.forClass(TransportProtos.SubscriptionInfoProto.class); |
||||
|
ArgumentCaptor<TransportServiceCallback<Void>> callbackCaptor = ArgumentCaptor.forClass(TransportServiceCallback.class); |
||||
|
|
||||
|
verify(transportServiceMock).process(sessionInfoCaptor.capture(), subscriptionInfoCaptor.capture(), callbackCaptor.capture()); |
||||
|
|
||||
|
assertThat(sessionInfoCaptor.getValue()).isEqualTo(expectedSessionInfo); |
||||
|
|
||||
|
TransportProtos.SubscriptionInfoProto expectedSubscriptionInfo = TransportProtos.SubscriptionInfoProto.newBuilder() |
||||
|
.setAttributeSubscription(expectedAttributesSubscription) |
||||
|
.setRpcSubscription(expectedRPCSubscription) |
||||
|
.setLastActivityTime(expectedTime) |
||||
|
.build(); |
||||
|
assertThat(subscriptionInfoCaptor.getValue()).isEqualTo(expectedSubscriptionInfo); |
||||
|
|
||||
|
TransportServiceCallback<Void> queueCallback = callbackCaptor.getValue(); |
||||
|
|
||||
|
queueCallback.onSuccess(null); |
||||
|
verify(callbackMock).onSuccess(SESSION_ID, expectedTime); |
||||
|
|
||||
|
var throwable = new Throwable(); |
||||
|
queueCallback.onError(throwable); |
||||
|
verify(callbackMock).onFailure(SESSION_ID, throwable); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenKeyAndTimeToReportAndSessionDoesNotExist_whenReportingActivity_thenShouldReportActivityWithNoSubscriptionsAndPreviousSessionInfo() { |
||||
|
// GIVEN
|
||||
|
long expectedTime = 123L; |
||||
|
boolean expectedAttributesSubscription = false; |
||||
|
boolean expectedRPCSubscription = false; |
||||
|
TransportProtos.SessionInfoProto expectedSessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
|
||||
|
ActivityReportCallback<UUID> callbackMock = mock(ActivityReportCallback.class); |
||||
|
|
||||
|
doCallRealMethod().when(transportServiceMock).reportActivity(SESSION_ID, expectedSessionInfo, expectedTime, callbackMock); |
||||
|
|
||||
|
// WHEN
|
||||
|
transportServiceMock.reportActivity(SESSION_ID, expectedSessionInfo, expectedTime, callbackMock); |
||||
|
|
||||
|
// THEN
|
||||
|
ArgumentCaptor<TransportProtos.SessionInfoProto> sessionInfoCaptor = ArgumentCaptor.forClass(TransportProtos.SessionInfoProto.class); |
||||
|
ArgumentCaptor<TransportProtos.SubscriptionInfoProto> subscriptionInfoCaptor = ArgumentCaptor.forClass(TransportProtos.SubscriptionInfoProto.class); |
||||
|
ArgumentCaptor<TransportServiceCallback<Void>> callbackCaptor = ArgumentCaptor.forClass(TransportServiceCallback.class); |
||||
|
|
||||
|
verify(transportServiceMock).process(sessionInfoCaptor.capture(), subscriptionInfoCaptor.capture(), callbackCaptor.capture()); |
||||
|
|
||||
|
assertThat(sessionInfoCaptor.getValue()).isEqualTo(expectedSessionInfo); |
||||
|
|
||||
|
TransportProtos.SubscriptionInfoProto expectedSubscriptionInfo = TransportProtos.SubscriptionInfoProto.newBuilder() |
||||
|
.setAttributeSubscription(expectedAttributesSubscription) |
||||
|
.setRpcSubscription(expectedRPCSubscription) |
||||
|
.setLastActivityTime(expectedTime) |
||||
|
.build(); |
||||
|
assertThat(subscriptionInfoCaptor.getValue()).isEqualTo(expectedSubscriptionInfo); |
||||
|
|
||||
|
TransportServiceCallback<Void> queueCallback = callbackCaptor.getValue(); |
||||
|
|
||||
|
queueCallback.onSuccess(null); |
||||
|
verify(callbackMock).onSuccess(SESSION_ID, expectedTime); |
||||
|
|
||||
|
var throwable = new Throwable(); |
||||
|
queueCallback.onError(throwable); |
||||
|
verify(callbackMock).onFailure(SESSION_ID, throwable); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenActivityHappened_whenRecordActivity_thenShouldDelegateToOnActivity() { |
||||
|
// GIVEN
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
doCallRealMethod().when(transportServiceMock).recordActivity(sessionInfo); |
||||
|
when(transportServiceMock.toSessionId(sessionInfo)).thenReturn(SESSION_ID); |
||||
|
long expectedTime = 123L; |
||||
|
when(transportServiceMock.getCurrentTimeMillis()).thenReturn(expectedTime); |
||||
|
|
||||
|
// WHEN
|
||||
|
transportServiceMock.recordActivity(sessionInfo); |
||||
|
|
||||
|
// THEN
|
||||
|
verify(transportServiceMock).onActivity(SESSION_ID, sessionInfo, expectedTime); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@EnumSource(ActivityStrategyType.class) |
||||
|
void givenDifferentReportingStrategies_whenGettingStrategy_thenShouldReturnCorrectStrategy(ActivityStrategyType reportingStrategyType) { |
||||
|
// GIVEN
|
||||
|
doCallRealMethod().when(transportServiceMock).getStrategy(); |
||||
|
ReflectionTestUtils.setField(transportServiceMock, "reportingStrategyType", reportingStrategyType); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityStrategy actualStrategy = transportServiceMock.getStrategy(); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(actualStrategy).isEqualTo(reportingStrategyType.toStrategy()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenSessionDoesNotExist_whenUpdatingActivityState_thenShouldReturnNull() { |
||||
|
// GIVEN
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>(); |
||||
|
state.setLastRecordedTime(123L); |
||||
|
state.setMetadata(sessionInfo); |
||||
|
|
||||
|
when(transportServiceMock.updateState(SESSION_ID, state)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> updatedState = transportServiceMock.updateState(SESSION_ID, state); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(updatedState).isNull(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenNoGwSessionId_whenUpdatingActivityState_thenShouldReturnSameInstanceWithUpdatedSessionInfo() { |
||||
|
// GIVEN
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(SESSION_ID, new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listenerMock)); |
||||
|
|
||||
|
long lastRecordedTime = 123L; |
||||
|
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>(); |
||||
|
state.setLastRecordedTime(lastRecordedTime); |
||||
|
state.setMetadata(TransportProtos.SessionInfoProto.getDefaultInstance()); |
||||
|
|
||||
|
when(transportServiceMock.updateState(SESSION_ID, state)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> updatedState = transportServiceMock.updateState(SESSION_ID, state); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(updatedState).isSameAs(state); |
||||
|
assertThat(updatedState.getLastRecordedTime()).isEqualTo(lastRecordedTime); |
||||
|
assertThat(updatedState.getMetadata()).isEqualTo(sessionInfo); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenHasGwSessionIdButGwSessionIsNotNull_whenUpdatingActivityState_thenShouldReturnSameInstanceWithUpdatedSessionInfo() { |
||||
|
// GIVEN
|
||||
|
var gwSessionId = UUID.fromString("19864038-9b48-11ee-b9d1-0242ac120002"); |
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.setGwSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setGwSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(SESSION_ID, new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listenerMock)); |
||||
|
|
||||
|
long lastRecordedTime = 123L; |
||||
|
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>(); |
||||
|
state.setLastRecordedTime(lastRecordedTime); |
||||
|
state.setMetadata(TransportProtos.SessionInfoProto.getDefaultInstance()); |
||||
|
|
||||
|
when(transportServiceMock.updateState(SESSION_ID, state)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> updatedState = transportServiceMock.updateState(SESSION_ID, state); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(updatedState).isSameAs(state); |
||||
|
assertThat(updatedState.getLastRecordedTime()).isEqualTo(lastRecordedTime); |
||||
|
assertThat(updatedState.getMetadata()).isEqualTo(sessionInfo); |
||||
|
|
||||
|
verify(transportServiceMock, never()).getLastRecordedTime(gwSessionId); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenHasGwSessionWithoutOverwriteEnabled_whenUpdatingActivityState_thenShouldReturnSameInstanceWithUpdatedSessionInfo() { |
||||
|
// GIVEN
|
||||
|
var gwSessionId = UUID.fromString("19864038-9b48-11ee-b9d1-0242ac120002"); |
||||
|
TransportProtos.SessionInfoProto gwSessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener gwListenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(gwSessionId, new SessionMetaData(gwSessionInfo, TransportProtos.SessionType.ASYNC, gwListenerMock)); |
||||
|
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.setGwSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setGwSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(SESSION_ID, new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listenerMock)); |
||||
|
|
||||
|
long lastRecordedTime = 123L; |
||||
|
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>(); |
||||
|
state.setLastRecordedTime(lastRecordedTime); |
||||
|
state.setMetadata(TransportProtos.SessionInfoProto.getDefaultInstance()); |
||||
|
|
||||
|
when(transportServiceMock.updateState(SESSION_ID, state)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> updatedState = transportServiceMock.updateState(SESSION_ID, state); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(updatedState).isSameAs(state); |
||||
|
assertThat(updatedState.getLastRecordedTime()).isEqualTo(lastRecordedTime); |
||||
|
assertThat(updatedState.getMetadata()).isEqualTo(sessionInfo); |
||||
|
|
||||
|
verify(transportServiceMock, never()).getLastRecordedTime(gwSessionId); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenHasGwSessionWithOverwriteEnabledAndGwLastRecordedTimeIsGreater_whenUpdatingActivityState_thenShouldReturnSameInstanceWithUpdatedSessionInfoAndLastRecordedTime() { |
||||
|
// GIVEN
|
||||
|
var gwSessionId = UUID.fromString("19864038-9b48-11ee-b9d1-0242ac120002"); |
||||
|
TransportProtos.SessionInfoProto gwSessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener gwListenerMock = mock(SessionMsgListener.class); |
||||
|
SessionMetaData gwSession = new SessionMetaData(gwSessionInfo, TransportProtos.SessionType.ASYNC, gwListenerMock); |
||||
|
gwSession.setOverwriteActivityTime(true); |
||||
|
sessions.put(gwSessionId, gwSession); |
||||
|
|
||||
|
long gwLastRecordedTime = 500L; |
||||
|
when(transportServiceMock.getLastRecordedTime(gwSessionId)).thenReturn(gwLastRecordedTime); |
||||
|
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.setGwSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setGwSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(SESSION_ID, new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listenerMock)); |
||||
|
|
||||
|
long lastRecordedTime = 123L; |
||||
|
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>(); |
||||
|
state.setLastRecordedTime(lastRecordedTime); |
||||
|
state.setMetadata(TransportProtos.SessionInfoProto.getDefaultInstance()); |
||||
|
|
||||
|
when(transportServiceMock.updateState(SESSION_ID, state)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> updatedState = transportServiceMock.updateState(SESSION_ID, state); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(updatedState).isSameAs(state); |
||||
|
assertThat(updatedState.getLastRecordedTime()).isEqualTo(gwLastRecordedTime); |
||||
|
assertThat(updatedState.getMetadata()).isEqualTo(sessionInfo); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenHasGwSessionWithOverwriteEnabledAndGwLastRecordedTimeIsLess_whenUpdatingActivityState_thenShouldReturnSameInstanceWithUpdatedSessionInfoOnly() { |
||||
|
// GIVEN
|
||||
|
var gwSessionId = UUID.fromString("19864038-9b48-11ee-b9d1-0242ac120002"); |
||||
|
TransportProtos.SessionInfoProto gwSessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener gwListenerMock = mock(SessionMsgListener.class); |
||||
|
SessionMetaData gwSession = new SessionMetaData(gwSessionInfo, TransportProtos.SessionType.ASYNC, gwListenerMock); |
||||
|
gwSession.setOverwriteActivityTime(true); |
||||
|
sessions.put(gwSessionId, gwSession); |
||||
|
|
||||
|
long gwLastRecordedTime = 100L; |
||||
|
when(transportServiceMock.getLastRecordedTime(gwSessionId)).thenReturn(gwLastRecordedTime); |
||||
|
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.setGwSessionIdMSB(gwSessionId.getMostSignificantBits()) |
||||
|
.setGwSessionIdLSB(gwSessionId.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(SESSION_ID, new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listenerMock)); |
||||
|
|
||||
|
long lastRecordedTime = 123L; |
||||
|
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> state = new ActivityState<>(); |
||||
|
state.setLastRecordedTime(lastRecordedTime); |
||||
|
state.setMetadata(TransportProtos.SessionInfoProto.getDefaultInstance()); |
||||
|
|
||||
|
when(transportServiceMock.updateState(SESSION_ID, state)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
ActivityState<TransportProtos.SessionInfoProto> updatedState = transportServiceMock.updateState(SESSION_ID, state); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(updatedState).isSameAs(state); |
||||
|
assertThat(updatedState.getLastRecordedTime()).isEqualTo(lastRecordedTime); |
||||
|
assertThat(updatedState.getMetadata()).isEqualTo(sessionInfo); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@MethodSource("provideTestParamsForHasExpiredTrue") |
||||
|
public void givenExpiredLastRecordedTime_whenCheckingForExpiry_thenShouldReturnTrue(long currentTimeMillis, long lastRecordedTime, long sessionInactivityTimeout) { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(transportServiceMock, "sessionInactivityTimeout", sessionInactivityTimeout); |
||||
|
|
||||
|
when(transportServiceMock.getCurrentTimeMillis()).thenReturn(currentTimeMillis); |
||||
|
when(transportServiceMock.hasExpired(lastRecordedTime)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
boolean hasExpired = transportServiceMock.hasExpired(lastRecordedTime); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(hasExpired).isTrue(); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> provideTestParamsForHasExpiredTrue() { |
||||
|
return Stream.of( |
||||
|
Arguments.of(10L, 0L, 9L), |
||||
|
Arguments.of(10L, 7L, 2L), |
||||
|
Arguments.of(10L, 8L, 1L), |
||||
|
Arguments.of(10000L, 5000L, 3000L) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@ParameterizedTest |
||||
|
@MethodSource("provideTestParamsForHasExpiredFalse") |
||||
|
public void givenNotExpiredLastRecordedTime_whenCheckingForExpiry_thenShouldReturnFalse(long currentTimeMillis, long lastRecordedTime, long sessionInactivityTimeout) { |
||||
|
// GIVEN
|
||||
|
ReflectionTestUtils.setField(transportServiceMock, "sessionInactivityTimeout", sessionInactivityTimeout); |
||||
|
|
||||
|
when(transportServiceMock.getCurrentTimeMillis()).thenReturn(currentTimeMillis); |
||||
|
when(transportServiceMock.hasExpired(lastRecordedTime)).thenCallRealMethod(); |
||||
|
|
||||
|
// WHEN
|
||||
|
boolean hasExpired = transportServiceMock.hasExpired(lastRecordedTime); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(hasExpired).isFalse(); |
||||
|
} |
||||
|
|
||||
|
private static Stream<Arguments> provideTestParamsForHasExpiredFalse() { |
||||
|
return Stream.of( |
||||
|
Arguments.of(10L, 9L, 2L), |
||||
|
Arguments.of(10L, 0L, 11L), |
||||
|
Arguments.of(10L, 8L, 3L), |
||||
|
Arguments.of(10000L, 8000L, 3000L) |
||||
|
); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenSessionExists_whenOnStateExpiryCalled_thenShouldPerformExpirationActions() { |
||||
|
// GIVEN
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
SessionMsgListener listenerMock = mock(SessionMsgListener.class); |
||||
|
sessions.put(SESSION_ID, new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listenerMock)); |
||||
|
doCallRealMethod().when(transportServiceMock).onStateExpiry(SESSION_ID, sessionInfo); |
||||
|
|
||||
|
// WHEN
|
||||
|
transportServiceMock.onStateExpiry(SESSION_ID, sessionInfo); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(sessions.containsKey(SESSION_ID)).isFalse(); |
||||
|
verify(transportServiceMock).deregisterSession(sessionInfo); |
||||
|
verify(transportServiceMock).process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); |
||||
|
verify(listenerMock).onRemoteSessionCloseCommand(SESSION_ID, SESSION_EXPIRED_NOTIFICATION_PROTO); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
void givenSessionDoesNotExist_whenOnStateExpiryCalled_thenShouldNotPerformExpirationActions() { |
||||
|
// GIVEN
|
||||
|
TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder() |
||||
|
.setSessionIdMSB(SESSION_ID.getMostSignificantBits()) |
||||
|
.setSessionIdLSB(SESSION_ID.getLeastSignificantBits()) |
||||
|
.build(); |
||||
|
doCallRealMethod().when(transportServiceMock).onStateExpiry(SESSION_ID, sessionInfo); |
||||
|
|
||||
|
// WHEN
|
||||
|
transportServiceMock.onStateExpiry(SESSION_ID, sessionInfo); |
||||
|
|
||||
|
// THEN
|
||||
|
assertThat(sessions.containsKey(SESSION_ID)).isFalse(); |
||||
|
verify(transportServiceMock, never()).deregisterSession(sessionInfo); |
||||
|
verify(transportServiceMock, never()).process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue