package com.dingzhuo.compute.engine.actor.alarm;
|
|
import static org.apache.commons.lang3.StringUtils.equalsAnyIgnoreCase;
|
|
import akka.actor.ActorSelection;
|
import akka.actor.UntypedAbstractActor;
|
import akka.event.Logging;
|
import akka.event.LoggingAdapter;
|
import com.dingzhuo.compute.engine.message.alarm.AlarmStatus;
|
import com.dingzhuo.compute.engine.message.alarm.LoadAlarmMessage;
|
import com.dingzhuo.compute.engine.message.alarm.UnloadAlarmMessage;
|
import com.dingzhuo.compute.engine.utils.ActorUtil;
|
import com.dingzhuo.compute.engine.utils.CacheService;
|
import com.dingzhuo.energy.common.utils.time.TimeType;
|
import com.dingzhuo.energy.data.model.domain.LimitType;
|
import com.dingzhuo.energy.data.model.service.ILimitTypeService;
|
import com.dingzhuo.energy.data.monitoring.alarm.domain.AlarmItem;
|
import com.dingzhuo.energy.data.monitoring.alarm.domain.RealTimeAlarm;
|
import com.dingzhuo.energy.data.monitoring.alarm.service.IAlarmItemService;
|
import com.dingzhuo.energy.data.monitoring.alarm.service.IRealtimeAlarmService;
|
import java.util.Date;
|
import java.util.HashMap;
|
import java.util.HashSet;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.concurrent.TimeUnit;
|
import java.util.stream.Collectors;
|
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
import org.springframework.context.annotation.Scope;
|
import org.springframework.stereotype.Component;
|
import scala.concurrent.duration.Duration;
|
|
/**
|
* @author fanxinfu
|
*/
|
@Component("loadAlarmActor")
|
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
public class LoadAlarmActor extends UntypedAbstractActor {
|
|
public static final String ACTOR_NAME = "loadAlarmActor";
|
LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
|
private final IAlarmItemService alarmItemService;
|
private final ILimitTypeService limitTypeService;
|
private ActorSelection realtimeAlarm;
|
private ActorSelection periodAlarm;
|
private CacheService cacheService;
|
private IRealtimeAlarmService realtimeAlarmService;
|
Map<String, AlarmItem> loadedAlarmItem = new HashMap<>();
|
|
public LoadAlarmActor(
|
IAlarmItemService alarmItemService,
|
ILimitTypeService limitTypeService,
|
IRealtimeAlarmService realtimeAlarmService,
|
CacheService cacheService) {
|
this.alarmItemService = alarmItemService;
|
this.limitTypeService = limitTypeService;
|
this.cacheService = cacheService;
|
this.realtimeAlarmService = realtimeAlarmService;
|
}
|
|
@Override
|
public void preStart() throws Exception {
|
super.preStart();
|
realtimeAlarm = getContext()
|
.actorSelection(ActorUtil.getActorAddress(RealtimeAlarmActor.ACTOR_NAME));
|
periodAlarm = getContext()
|
.actorSelection(ActorUtil.getActorAddress(PeriodAlarmActor.ACTOR_NAME));
|
this.context().system().scheduler()
|
.scheduleAtFixedRate(Duration.Zero(), Duration.create(1, TimeUnit.MINUTES), this.self(),
|
Message.REFRESH, this.context().system().dispatcher(), null);
|
cacheLimitType();
|
initAlarm();
|
}
|
|
private void cacheLimitType() {
|
List<LimitType> limitTypes = limitTypeService.selectLimitTypeList(new LimitType());
|
limitTypes.forEach(type -> cacheService.cacheLimitType(type.getLimitCode(), type));
|
}
|
|
private void initAlarm() {
|
List<RealTimeAlarm> alarms = realtimeAlarmService.getRealTimeAlarm();
|
if (!alarms.isEmpty()) {
|
alarms.forEach(alarm -> {
|
AlarmStatus status = new AlarmStatus();
|
status.setAlarm(true);
|
status.setBeginTime(alarm.getBeginTime());
|
cacheService.cacheAlarmStatus(alarm.getItemId(), status);
|
});
|
}
|
}
|
|
@Override
|
public void onReceive(Object message) {
|
if (message instanceof Message) {
|
if (message == Message.REFRESH) {
|
refreshAlarmItem();
|
} else {
|
this.unhandled(message);
|
}
|
}
|
}
|
|
private void refreshAlarmItem() {
|
List<AlarmItem> alarmItemList = alarmItemService.getAllAlarmItem();
|
Map<String, AlarmItem> newAlarmItem = alarmItemList.stream()
|
.collect(Collectors.toMap(ActorUtil::buildAlarmActorId, item -> item));
|
|
Set<String> needInstall = new HashSet<>();
|
Set<String> needUninstall = new HashSet<>();
|
|
loadedAlarmItem.forEach((id, alarmItem) -> {
|
if (!newAlarmItem.containsKey(id)) {
|
needUninstall.add(id);
|
} else {
|
Date nowUpdate = newAlarmItem.get(id).getUpdateTime();
|
Date lastUpdate = alarmItem.getUpdateTime();
|
if (lastUpdate != null && nowUpdate != null && lastUpdate.after(nowUpdate)) {
|
needUninstall.add(id);
|
needInstall.add(id);
|
}
|
}
|
});
|
|
newAlarmItem.forEach((id, alarmItem) -> {
|
if (!loadedAlarmItem.containsKey(id)) {
|
needInstall.add(id);
|
}
|
});
|
|
needUninstall.forEach(id -> {
|
AlarmItem alarmItem = loadedAlarmItem.get(id);
|
loadedAlarmItem.remove(id);
|
if (equalsAnyIgnoreCase(alarmItem.getTimeSlot(), TimeType.LIVE.name())) {
|
realtimeAlarm.tell(new UnloadAlarmMessage(alarmItem), getSelf());
|
} else {
|
periodAlarm.tell(new UnloadAlarmMessage(alarmItem), getSelf());
|
}
|
});
|
|
needInstall.forEach(id -> {
|
AlarmItem alarmItem = newAlarmItem.get(id);
|
if (equalsAnyIgnoreCase(alarmItem.getTimeSlot(), TimeType.LIVE.name())) {
|
alarmItem.setCalcText("limitRealtimeAlarm('" + id + "')");
|
realtimeAlarm.tell(new LoadAlarmMessage(alarmItem), getSelf());
|
} else {
|
alarmItem.setCalcText("limitPeriodAlarm('" + id + "')");
|
periodAlarm.tell(new LoadAlarmMessage(alarmItem), getSelf());
|
}
|
|
loadedAlarmItem.put(id, alarmItem);
|
});
|
}
|
|
public enum Message {
|
/**
|
* 检测报警配置
|
*/
|
REFRESH
|
}
|
}
|
