Agv Event 体系

servo/src/main/java/com/galaxis/rcs/common/enums/AgvEventType.java

@@ -1,10 +0,0 @@
-package com.galaxis.rcs.common.enums;
-
-public enum AgvEventType {
-    OFFLINE, ONLINE, FREE,
-    PLAN_COMPLETE, PLAN_ACCEPT, PLAN_CANCEL,
-    PLAN_PAUSE, PLAN_RESUME,
-    DEVICE_TASK_COMPLETE, DEVICE_TASK_EXCEPTION,
-    POS_CHANGED, DIRECTION_CHANGED,
-    LOW_BATTERY, STUCK
-}

servo/src/main/java/com/galaxis/rcs/ptr/AgvEventManager.java

@@ -1,66 +0,0 @@
-package com.galaxis.rcs.ptr;
-
-import com.galaxis.rcs.common.enums.AgvEventType;
-import com.yvan.logisticsModel.LogisticsRuntime;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.*;
-
-@Slf4j
-public class AgvEventManager {
-    private final Map<AgvEventType, List<AgvEventListener>> listeners = new ConcurrentHashMap<>();
-    private final Map<String, ScheduledFuture<?>> stuckMonitors = new ConcurrentHashMap<>();
-    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);
-    private final LogisticsRuntime runtime;
-
-    public AgvEventManager(LogisticsRuntime runtime) {
-        this.runtime = runtime;
-    }
-
-    public void registerListener(AgvEventType type, AgvEventListener listener) {
-        listeners.computeIfAbsent(type, k -> new CopyOnWriteArrayList<>()).add(listener);
-    }
-
-    public void unregisterListener(AgvEventType type, AgvEventListener listener) {
-        List<AgvEventListener> list = listeners.get(type);
-        if (list != null) {
-            list.remove(listener);
-        }
-    }
-
-    public void fireEvent(AgvEventType type, Object... args) {
-        List<AgvEventListener> list = listeners.get(type);
-        if (list != null) {
-            for (AgvEventListener listener : list) {
-                try {
-                    listener.onEvent(type, args);
-                } catch (Exception e) {
-                    log.error("Error handling event {}", type, e);
-                }
-            }
-        }
-    }
-
-    public void monitorMovement(String agvId, PosDirection startPos, int delayOfSeconds) {
-        // 取消现有的监控
-        cancelStuckMonitor(agvId);
-
-        ScheduledFuture<?> future = scheduler.schedule(() -> {
-            PtrAgvItem agv = (PtrAgvItem) this.runtime.executorItemMap.get(agvId);
-            if (agv != null && agv.isSamePosition(startPos)) {
-                fireEvent(AgvEventType.STUCK, agv, delayOfSeconds);
-            }
-        }, delayOfSeconds, TimeUnit.SECONDS);
-
-        stuckMonitors.put(agvId, future);
-    }
-
-    public void cancelStuckMonitor(String agvId) {
-        ScheduledFuture<?> future = stuckMonitors.remove(agvId);
-        if (future != null) {
-            future.cancel(false);
-        }
-    }
-}

servo/src/main/java/com/galaxis/rcs/ptr/PtrAgvItem.java

@@ -2,7 +2,7 @@ package com.galaxis.rcs.ptr;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.galaxis.rcs.common.entity.RcsTaskPlan;
-import com.galaxis.rcs.common.enums.AgvEventType;
+import com.yvan.event.AgvEventType;
 import com.galaxis.rcs.common.enums.LCCDirection;
 import com.galaxis.rcs.common.enums.PlanTaskStatus;
 import com.galaxis.rcs.common.enums.PlanTaskType;
@@ -15,6 +15,7 @@ import com.galaxis.rcs.ptr.sendEntity.RcsSRMessage;
 import com.galaxis.rcs.ptr.sendEntity.RcsSetLocationMessage;
 import com.google.common.base.Joiner;
 import com.google.common.collect.Queues;
+import com.yvan.event.EventManager;
 import com.yvan.logisticsModel.ExecutorItem;
 import com.yvan.logisticsModel.LogisticsRuntime;
 import com.yvan.logisticsModel.StaticItem;
@@ -73,8 +74,6 @@ public abstract class PtrAgvItem extends ExecutorItem {
     private volatile PosDirection lastPausedPosition;
     private volatile boolean isOnline = false;
 
-    private final Set<AgvEventListener> eventListeners = new CopyOnWriteArraySet<>();
-
     // 执行中的任务
     @JsonIgnore
     public List<PtrAgvDeviceTask> runningDeviceTaskList = new ArrayList<>();
@@ -91,6 +90,7 @@ public abstract class PtrAgvItem extends ExecutorItem {
     @JsonIgnore
     final BlockingQueue<PtrAgvDeviceTask> deviceTaskQueue = Queues.newArrayBlockingQueue(BLOCKING_QUEUE_CAPACITY);
 
+    @JsonIgnore
     final Cl2DeviceConnector cl2DeviceConnector = new Cl2DeviceConnector(this.runtime);
 
     @JsonIgnore
@@ -159,10 +159,14 @@ public abstract class PtrAgvItem extends ExecutorItem {
 
         this.planTaskSequence = taskSequence;
         buildPlanToDeviceTask();
-        fireEvent(AgvEventType.PLAN_ACCEPT, this);
+        fireEvent(AgvEventType.PLAN_TASK_SEQUENCE_ACCEPT, this, taskSequence);
         connectorThread.resumeProcessing();
     }
 
+    private void fireEvent(AgvEventType agvEventType, Object... args) {
+        this.runtime.eventManager.fireEvent(agvEventType, args);
+    }
+
     public synchronized void pauseTask() {
         if (planTaskSequence == null) {
             throw new IllegalStateException("No active task to pause");
@@ -184,7 +188,7 @@ public abstract class PtrAgvItem extends ExecutorItem {
             log.error("Failed to send stop command to AGV {}", this.getId(), e);
         }
 
-        fireEvent(AgvEventType.PLAN_PAUSE, this);
+        fireEvent(AgvEventType.PLAN_TASK_SEQUENCE_PAUSE, this);
     }
 
     public synchronized void resumeTask() {
@@ -202,7 +206,7 @@ public abstract class PtrAgvItem extends ExecutorItem {
 
         isPaused = false;
         connectorThread.resumeProcessing();
-        fireEvent(AgvEventType.PLAN_RESUME, this);
+        fireEvent(AgvEventType.PLAN_TASK_SEQUENCE_RESUME, this);
     }
 
     @SneakyThrows
@@ -217,7 +221,7 @@ public abstract class PtrAgvItem extends ExecutorItem {
             deviceTaskQueue.clear();
         }
 
-        fireEvent(AgvEventType.PLAN_CANCEL, this);
+        fireEvent(AgvEventType.PLAN_TASK_SEQUENCE_CANCEL, this);
     }
 
     @SneakyThrows
@@ -284,7 +288,7 @@ public abstract class PtrAgvItem extends ExecutorItem {
         }
 
         if (planTaskSequence != null && planTaskSequence.isAllCompleted()) {
-            fireEvent(AgvEventType.PLAN_COMPLETE, this);
+            fireEvent(AgvEventType.PLAN_TASK_SEQUENCE_COMPLETE, this);
             this.runningDeviceTaskList.clear();
             planTaskSequence = null;
         }
@@ -306,13 +310,12 @@ public abstract class PtrAgvItem extends ExecutorItem {
         // 触发位置变化事件
         if (oldX != logicX || oldY != logicY) {
             fireEvent(AgvEventType.POS_CHANGED, this,
-                new PosDirection(oldX, oldY, oldDirection),
+                new PosDirection(logicX, logicY, direction),
-                new PosDirection(logicX, logicY, direction));
+                new PosDirection(oldX, oldY, oldDirection));
         }
 
         if (oldDirection != direction) {
-            fireEvent(AgvEventType.DIRECTION_CHANGED, this,
+            fireEvent(AgvEventType.DIRECTION_CHANGED, this, direction, oldDirection);
-                oldDirection, direction);
         }
 
         boolean needCompute = false;
@@ -341,14 +344,15 @@ public abstract class PtrAgvItem extends ExecutorItem {
 
                     // 更新计划任务
                     RcsTaskPlan planTask = planTaskSequence.getByPlanTaskId(task.movePlanTaskId);
-                    if (planTask != null) {
+                    if (planTask != null && !PlanTaskStatus.FINISHED.toString().equals(planTask.getPlanTaskStatus())) {
+                        fireEvent(AgvEventType.DEVICE_TASK_COMPLETE, this, planTaskSequence, planTask);
                         planTask.setPlanTaskStatus(PlanTaskStatus.FINISHED.toString());
                         planTaskSequence.savePlanTask(planTask);
                     }
                 }
 
                 if (planTaskSequence.isAllCompleted()) {
-                    fireEvent(AgvEventType.PLAN_COMPLETE, this);
+                    fireEvent(AgvEventType.PLAN_TASK_SEQUENCE_COMPLETE, this);
                     this.runningDeviceTaskList.clear();
                     planTaskSequence = null;
                 }
@@ -446,21 +450,6 @@ public abstract class PtrAgvItem extends ExecutorItem {
         fireEvent(AgvEventType.OFFLINE, this);
     }
 
-    // 事件监听管理
-    public void addEventListener(AgvEventListener listener) {
-        eventListeners.add(listener);
-    }
-
-    public void removeEventListener(AgvEventListener listener) {
-        eventListeners.remove(listener);
-    }
-
-    private void fireEvent(AgvEventType type, Object... args) {
-        for (AgvEventListener listener : eventListeners) {
-            listener.onEvent(type, args);
-        }
-    }
-
     private String getTaskStatus() {
         if (planTaskSequence == null) return "IDLE";
         if (isPaused) return "PAUSED";
@@ -801,11 +790,6 @@ public abstract class PtrAgvItem extends ExecutorItem {
 
 //        planQueue.addAll(sequence.taskList);
         deviceTaskQueue.addAll(deviceTaskList);
-
-        String json = JsonWrapper.toJson(deviceTaskList);
-        log.info("deviceTaskList: {}", json);
-
-        // TODO: 开启轮询线程，等待下一个待执行任务
     }
 
     public boolean isSamePosition(PosDirection startPos) {

servo/src/main/java/com/galaxis/rcs/ptr/AgvEventListener.java → servo/src/main/java/com/yvan/event/AgvEventListener.java

@@ -1,6 +1,4 @@
-package com.galaxis.rcs.ptr;
+package com.yvan.event;
-
-import com.galaxis.rcs.common.enums.AgvEventType;
 
 public interface AgvEventListener {
     void onEvent(AgvEventType type, Object... args);

servo/src/main/java/com/yvan/event/AgvEventType.java

@@ -0,0 +1,82 @@
+package com.yvan.event;
+
+public enum AgvEventType {
+    /**
+     * 设备离线
+     */
+    OFFLINE,
+    /**
+     * 设备上线
+     */
+    ONLINE,
+    /**
+     * 设备阻挡
+     */
+    BLOCKED,
+    /**
+     * 设备阻挡恢复
+     */
+    BLOCKED_RECOVER,
+    /**
+     * 设备空闲
+     */
+    FREE,
+    /**
+     * 规划任务序列全部完成
+     * (ExecutorItem sender)
+     */
+    PLAN_TASK_SEQUENCE_COMPLETE,
+    /**
+     * 某一个规划任务被完成
+     * (ExecutorItem sender, PlanTaskSequence taskSequence, RcsTaskPlan taskPlan)
+     */
+    PLAN_TASK_COMPLETE,
+    /**
+     * 规划任务异常
+     */
+    PLAN_TASK_EXCEPTION,
+    /**
+     * 规划任务已接受
+     * (ExecutorItem sender, PlanTaskSequence taskSequence)
+     */
+    PLAN_TASK_SEQUENCE_ACCEPT,
+    /**
+     * 规划任务已取消
+     * (ExecutorItem sender)
+     */
+    PLAN_TASK_SEQUENCE_CANCEL,
+    /**
+     * 规划任务已暂停
+     * (ExecutorItem sender)
+     */
+    PLAN_TASK_SEQUENCE_PAUSE,
+    /**
+     * 规划任务已恢复
+     * (ExecutorItem sender)
+     */
+    PLAN_TASK_SEQUENCE_RESUME,
+    /**
+     * 设备任务已完成
+     * (ExecutorItem sender, PtrAgvDeviceTask deviceTask)
+     */
+    DEVICE_TASK_COMPLETE,
+    /**
+     * 设备任务异常
+     */
+    DEVICE_TASK_EXCEPTION,
+    /**
+     * 设备位置改变
+     * (ExecutorItem sender, PosDirection newPosDirection, PosDirection originPosDirection)
+     */
+    POS_CHANGED,
+    /**
+     * 设备姿态改变
+     * (ExecutorItem sender, PosDirection newPosDirection, PosDirection originPosDirection)
+     */
+    DIRECTION_CHANGED,
+    /**
+     * 设备低电量
+     * (ExecutorItem sender)
+     */
+    LOW_BATTERY,
+}

servo/src/main/java/com/yvan/event/EventManager.java

@@ -0,0 +1,24 @@
+package com.yvan.event;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+public class EventManager {
+
+    private final Set<AgvEventListener> eventListeners = new CopyOnWriteArraySet<>();
+
+    public void fireEvent(AgvEventType type, Object... args) {
+        for (AgvEventListener listener : eventListeners) {
+            listener.onEvent(type, args);
+        }
+    }
+
+    // 事件监听管理
+    public void subscribe(AgvEventListener listener) {
+        eventListeners.add(listener);
+    }
+
+    public void unsubscribe(AgvEventListener listener) {
+        eventListeners.remove(listener);
+    }
+}

servo/src/main/java/com/yvan/logisticsModel/LogisticsRuntime.java

@@ -1,10 +1,8 @@
 package com.yvan.logisticsModel;
 
-import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.galaxis.rcs.connector.cl2.Cl2Item;
 import com.galaxis.rcs.plan.path.NavigationGraph;
 import com.galaxis.rcs.plan.path.PtrPathPlanner;
-import com.galaxis.rcs.ptr.AgvEventManager;
 import com.galaxis.rcs.ptr.AmrMessageHandler;
 import com.galaxis.rcs.task.TaskDispatchFactory;
 import com.galaxis.rcs.task.TaskService;
@@ -12,7 +10,8 @@ import com.google.common.base.Joiner;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
-import com.yvan.mqtt.LccMqttService;
+import com.yvan.event.EventManager;
+import com.yvan.mqtt.FrontendMessagePushService;
 import com.yvan.redis.LccRedisService;
 import com.yvan.workbench.model.entity.LccProject;
 import com.yvan.workbench.model.entity.LccProjectEnv;
@@ -28,7 +27,7 @@ import java.util.Map;
 import java.util.Set;
 
 /**
- * 物流上下文运行时
+ * 物流运行时
  */
 @Slf4j
 public class LogisticsRuntime {
@@ -85,13 +84,11 @@ public class LogisticsRuntime {
     /**
      * 任务服务
      */
-    @JsonIgnore
     public final TaskService taskService = new TaskService(this);
 
     /**
      * 任务分配服务
      */
-    @JsonIgnore
     public final TaskDispatchFactory taskDispatchFactory = new TaskDispatchFactory(this);
 
     /**
@@ -102,7 +99,6 @@ public class LogisticsRuntime {
     /**
      * AGV导航地图 车类型 t -> NavigationGraph
      */
-    @JsonIgnore
     public final Map<String, PtrPathPlanner> pathPlannerMap = Maps.newHashMap();
 
     /**
@@ -110,15 +106,14 @@ public class LogisticsRuntime {
      */
     public final Map<String, Floor> floorMap = Maps.newHashMap();
 
-    @JsonIgnore
     public final AmrMessageHandler amrMessageHandler = new AmrMessageHandler(this);
 
-    public final AgvEventManager agvEventManager = new AgvEventManager(this);
+    public final FrontendMessagePushService frontendMessagePushService = new FrontendMessagePushService(this);
-
-    public final LccMqttService lccMqttService = new LccMqttService(this);
 
     public final LccRedisService lccRedisService = new LccRedisService(this);
 
+    public final EventManager eventManager = new EventManager();
+
     public LogisticsRuntime(LccProject project, LccProjectEnv env, String serverId) {
         this.project = project;
         this.env = env;
@@ -244,7 +239,7 @@ public class LogisticsRuntime {
 
         // 启动 MQTT 监听
         this.amrMessageHandler.start(this.env.getEnvConfig().getMqtt(), this.serverId);
-        this.lccMqttService.start(this.env.getEnvConfig().getMqtt(), this.serverId + "_lcc_send");
+        this.frontendMessagePushService.start(this.env.getEnvConfig().getMqtt(), this.serverId + "_lcc_send");
         this.lccRedisService.start(this.env.getEnvConfig().getRedis(), this.serverId);
 
         // 开启所有机器人的任务处理
@@ -314,7 +309,7 @@ public class LogisticsRuntime {
 
         // 停止 MQTT 监听
         this.amrMessageHandler.stop();
-        this.lccMqttService.stop();
+        this.frontendMessagePushService.stop();
         this.lccRedisService.stop();
 
         BannerUtils.printConfig(log, "LogisticsRuntime stop.", new String[]{

servo/src/main/java/com/yvan/mqtt/LccMqttService.java → servo/src/main/java/com/yvan/mqtt/FrontendMessagePushService.java

@@ -21,7 +21,7 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 @Slf4j
-public class LccMqttService implements MqttCallback {
+public class FrontendMessagePushService implements MqttCallback {
     private volatile EnvConfig.MqttConfig mqttConfig;
     private volatile String clientId;
     private volatile MqttClient mqttClient;
@@ -41,7 +41,7 @@ public class LccMqttService implements MqttCallback {
     @Getter
     private volatile boolean connected = false;
 
-    public LccMqttService(LogisticsRuntime runtime) {
+    public FrontendMessagePushService(LogisticsRuntime runtime) {
         this.runtime = runtime;
     }
 

servo/src/main/java/com/yvan/state/VariableMonitor.java

@@ -0,0 +1,114 @@
+package com.yvan.state;
+
+import org.clever.core.mapper.JacksonMapper;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/**
+ * VariableMonitor<T> 类用于周期性检查某个变量的值是否发生变化。
+ * 如果发生变化，就触发回调通知所有监听器。
+ * <p>
+ * 本实现使用 Jackson 序列化对象为 JSON 字符串进行比较，适用于：
+ * - Map
+ * - List
+ * - Set
+ * - 自定义实体类
+ * - 嵌套结构
+ * <p>
+ * 并自动防止循环引用问题。
+ *
+ * @param <T> 被监控变量的类型
+ */
+public class VariableMonitor<T> {
+
+    // 提供当前变量值的方法
+    private final Supplier<T> provider;
+
+    // 轮询间隔时间（毫秒）
+    private final long intervalMillis;
+
+    // 定时任务执行器
+    private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
+    // 当前调度的任务对象
+    private ScheduledFuture<?> future;
+
+    // 上一次变量的 JSON 表示
+    private String lastJson;
+
+    // 存储所有的监听器
+    private final List<Consumer<T>> listeners = new ArrayList<>();
+
+    public VariableMonitor(Supplier<T> provider, long intervalMillis) {
+        this.provider = provider;
+        this.intervalMillis = intervalMillis;
+        this.lastJson = JacksonMapper.getInstance().toJson(provider.get());
+    }
+
+    /**
+     * 启动定时轮询任务
+     */
+    public void start() {
+        future = scheduler.scheduleAtFixedRate(() -> {
+            T currentValue = provider.get();
+            String currentJson;
+            currentJson = JacksonMapper.getInstance().toJson(currentValue);
+
+            if (!Objects.equals(lastJson, currentJson)) {
+                notifyListeners(currentValue);
+                lastJson = currentJson;
+            }
+        }, 0, intervalMillis, TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * 停止定时任务
+     */
+    public void stop() {
+        if (future != null) {
+            future.cancel(false);
+        }
+        scheduler.shutdownNow();
+    }
+
+    /**
+     * 添加一个监听器
+     *
+     * @param listener 回调函数，接收新值作为参数
+     */
+    public void addListener(Consumer<T> listener) {
+        listeners.add(listener);
+    }
+
+    /**
+     * 移除一个监听器
+     *
+     * @param listener 要移除的监听器
+     */
+    public void removeListener(Consumer<T> listener) {
+        listeners.remove(listener);
+    }
+
+    /**
+     * 通知所有监听器变量发生了变化
+     *
+     * @param newValue 新的变量值
+     */
+    private void notifyListeners(T newValue) {
+        for (Consumer<T> listener : listeners) {
+            try {
+                listener.accept(newValue);
+            } catch (Exception e) {
+                System.err.println("监听器执行出错: " + e.getMessage());
+            }
+        }
+    }
+}