266 lines
11 KiB
Java
266 lines
11 KiB
Java
package com.dongni.collisionavoidance.geofence.event;
|
||
|
||
import com.dongni.collisionavoidance.geofence.model.entity.RuleViolationEvent;
|
||
import com.dongni.collisionavoidance.geofence.model.entity.SpatialRule;
|
||
import com.dongni.collisionavoidance.geofence.model.enums.RuleExecutionResult;
|
||
import com.dongni.collisionavoidance.geofence.model.enums.RuleStatus;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.beans.factory.annotation.Autowired;
|
||
import org.springframework.context.ApplicationEventPublisher;
|
||
import org.springframework.stereotype.Component;
|
||
|
||
import java.time.LocalDateTime;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
|
||
/**
|
||
* 规则事件发布器
|
||
* 负责统一发布各种规则相关事件
|
||
*/
|
||
@Slf4j
|
||
@Component
|
||
public class RuleEventPublisher {
|
||
|
||
@Autowired
|
||
private ApplicationEventPublisher eventPublisher;
|
||
|
||
/**
|
||
* 发布规则违规事件
|
||
*
|
||
* @param violationEvent 违规事件详情
|
||
* @param source 事件源对象
|
||
* @param contextInfo 上下文信息
|
||
*/
|
||
public void publishViolationEvent(RuleViolationEvent violationEvent, Object source, String contextInfo) {
|
||
try {
|
||
RuleViolationEventOccurred event = new RuleViolationEventOccurred(source, violationEvent,
|
||
"RuleEngine", contextInfo);
|
||
|
||
log.info("发布规则违规事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
// 记录事件发布统计
|
||
recordEventPublication("VIOLATION", event.getPriority().name());
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则违规事件失败: violationEventId={}, error={}",
|
||
violationEvent.getEventId(), e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
// 异步发布方法已移除 - 简化架构，统一使用同步发布
|
||
|
||
/**
|
||
* 批量发布规则违规事件
|
||
*/
|
||
public void publishBatchViolationEvents(List<RuleViolationEvent> violationEvents, Object source) {
|
||
if (violationEvents == null || violationEvents.isEmpty()) {
|
||
return;
|
||
}
|
||
|
||
log.info("批量发布规则违规事件，数量: {}", violationEvents.size());
|
||
|
||
violationEvents.forEach(violationEvent -> {
|
||
publishViolationEvent(violationEvent, source,
|
||
String.format("批量违规检测 - 共%d个事件", violationEvents.size()));
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 发布规则执行开始事件
|
||
*/
|
||
public void publishRuleExecutionStarted(SpatialRule rule, String vehicleId, Object source,
|
||
Map<String, Object> executionContext) {
|
||
try {
|
||
RuleExecutionEvent event = RuleExecutionEvent.executionStarted(source, rule, vehicleId,
|
||
"RuleExecutionEngine", executionContext);
|
||
|
||
log.debug("发布规则执行开始事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("EXECUTION_STARTED", "NORMAL");
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则执行开始事件失败: ruleId={}, vehicleId={}, error={}",
|
||
rule.getRuleId(), vehicleId, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发布规则执行完成事件
|
||
*/
|
||
public void publishRuleExecutionCompleted(SpatialRule rule, String vehicleId, RuleExecutionResult result,
|
||
LocalDateTime startTime, Object source,
|
||
Map<String, Object> executionContext, String details) {
|
||
try {
|
||
RuleExecutionEvent event = RuleExecutionEvent.executionCompleted(source, rule, vehicleId,
|
||
result, startTime, "RuleExecutionEngine", executionContext, details);
|
||
|
||
log.debug("发布规则执行完成事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("EXECUTION_COMPLETED", result.name());
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则执行完成事件失败: ruleId={}, vehicleId={}, result={}, error={}",
|
||
rule.getRuleId(), vehicleId, result.name(), e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发布规则执行失败事件
|
||
*/
|
||
public void publishRuleExecutionFailed(SpatialRule rule, String vehicleId, LocalDateTime startTime,
|
||
Object source, Map<String, Object> executionContext,
|
||
String errorMessage) {
|
||
try {
|
||
RuleExecutionEvent event = RuleExecutionEvent.executionFailed(source, rule, vehicleId,
|
||
startTime, "RuleExecutionEngine", executionContext, errorMessage);
|
||
|
||
log.warn("发布规则执行失败事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("EXECUTION_FAILED", "ERROR");
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则执行失败事件失败: ruleId={}, vehicleId={}, error={}",
|
||
rule.getRuleId(), vehicleId, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发布规则状态变更事件
|
||
*/
|
||
public void publishRuleStateChange(SpatialRule rule, RuleStatus previousStatus, RuleStatus newStatus,
|
||
String changeReason, String changedBy, Object source,
|
||
Map<String, Object> changeContext) {
|
||
try {
|
||
RuleStateChangeEvent event = new RuleStateChangeEvent(source, rule, previousStatus, newStatus,
|
||
changeReason, changedBy, "RuleManager", changeContext,
|
||
RuleStateChangeEvent.StateChangeType.MANUAL_CHANGE);
|
||
|
||
log.info("发布规则状态变更事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("STATE_CHANGE", event.getChangeType().name());
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则状态变更事件失败: ruleId={}, previousStatus={}, newStatus={}, error={}",
|
||
rule.getRuleId(), previousStatus, newStatus, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发布规则激活事件
|
||
*/
|
||
public void publishRuleActivated(SpatialRule rule, String changedBy, String reason, Object source) {
|
||
try {
|
||
RuleStateChangeEvent event = RuleStateChangeEvent.ruleActivated(source, rule, changedBy, reason);
|
||
|
||
log.info("发布规则激活事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("RULE_ACTIVATED", "HIGH");
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则激活事件失败: ruleId={}, changedBy={}, error={}",
|
||
rule.getRuleId(), changedBy, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发布规则停用事件
|
||
*/
|
||
public void publishRuleDeactivated(SpatialRule rule, String changedBy, String reason, Object source) {
|
||
try {
|
||
RuleStateChangeEvent event = RuleStateChangeEvent.ruleDeactivated(source, rule, changedBy, reason);
|
||
|
||
log.info("发布规则停用事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("RULE_DEACTIVATED", "HIGH");
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则停用事件失败: ruleId={}, changedBy={}, error={}",
|
||
rule.getRuleId(), changedBy, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 发布规则配置更新事件
|
||
*/
|
||
public void publishRuleConfigUpdated(SpatialRule rule, String changedBy, Map<String, Object> updateContext,
|
||
Object source) {
|
||
try {
|
||
RuleStateChangeEvent event = RuleStateChangeEvent.ruleConfigUpdated(source, rule, changedBy, updateContext);
|
||
|
||
log.info("发布规则配置更新事件: {}", event.getEventDescription());
|
||
eventPublisher.publishEvent(event);
|
||
|
||
recordEventPublication("CONFIG_UPDATED", "NORMAL");
|
||
|
||
} catch (Exception e) {
|
||
log.error("发布规则配置更新事件失败: ruleId={}, changedBy={}, error={}",
|
||
rule.getRuleId(), changedBy, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 批量发布规则状态变更事件
|
||
*/
|
||
public void publishBatchRuleStateChanges(List<SpatialRule> rules, RuleStatus newStatus,
|
||
String changeReason, String changedBy, Object source) {
|
||
if (rules == null || rules.isEmpty()) {
|
||
return;
|
||
}
|
||
|
||
log.info("批量发布规则状态变更事件，数量: {}, 新状态: {}", rules.size(), newStatus);
|
||
|
||
rules.forEach(rule -> {
|
||
publishRuleStateChange(rule, rule.getStatus(), newStatus, changeReason, changedBy, source, null);
|
||
});
|
||
}
|
||
|
||
// 异步发布方法已移除 - 简化架构，统一使用同步发布
|
||
|
||
/**
|
||
* 检查事件发布是否启用
|
||
*/
|
||
public boolean isEventPublishingEnabled() {
|
||
return eventPublisher != null;
|
||
}
|
||
|
||
/**
|
||
* 获取事件发布统计信息
|
||
*/
|
||
public Map<String, Long> getEventPublishingStatistics() {
|
||
// 这里可以集成实际的统计存储，暂时返回空Map
|
||
return Map.of(
|
||
"totalEventsPublished", 0L,
|
||
"violationEvents", 0L,
|
||
"executionEvents", 0L,
|
||
"stateChangeEvents", 0L
|
||
);
|
||
}
|
||
|
||
/**
|
||
* 记录事件发布统计
|
||
*/
|
||
private void recordEventPublication(String eventType, String eventCategory) {
|
||
// 这里可以集成实际的统计记录逻辑
|
||
log.debug("记录事件发布统计: type={}, category={}", eventType, eventCategory);
|
||
}
|
||
|
||
/**
|
||
* 紧急事件发布（用于高优先级事件）
|
||
*/
|
||
public void publishEmergencyEvent(RuleViolationEvent violationEvent, Object source) {
|
||
// 紧急事件同步发布，确保立即处理
|
||
publishViolationEvent(violationEvent, source, "紧急违规事件");
|
||
|
||
// 额外的紧急处理逻辑
|
||
log.error("紧急规则违规事件已发布: vehicleId={}, ruleId={}, violationType={}",
|
||
violationEvent.getVehicleId(), violationEvent.getRuleId(),
|
||
violationEvent.getViolationType());
|
||
}
|
||
} |