|
@@ -78,6 +78,9 @@ type Router struct {
|
|
|
|
|
|
|
|
var routers = make(map[string]*Router)
|
|
var routers = make(map[string]*Router)
|
|
|
|
|
|
|
|
|
|
+// Per-router command queue (for offline routers - auto-flush on reconnect)
|
|
|
|
|
+var routerQueues = make(map[string][]RouterCommand)
|
|
|
|
|
+
|
|
|
// Pending commands awaiting results (command_id -> result channel)
|
|
// Pending commands awaiting results (command_id -> result channel)
|
|
|
var pendingCommands = make(map[string]chan CommandResult)
|
|
var pendingCommands = make(map[string]chan CommandResult)
|
|
|
const commandTimeout = 30 * time.Second
|
|
const commandTimeout = 30 * time.Second
|
|
@@ -187,7 +190,21 @@ func handleWebSocket(w http.ResponseWriter, r *http.Request) {
|
|
|
Connected: true,
|
|
Connected: true,
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- log.Printf("Router connected: %s", routerID)
|
|
|
|
|
|
|
+ log.Printf("Router connected: %s (flushing queue)", routerID)
|
|
|
|
|
+
|
|
|
|
|
+ // Flush queued commands for this router
|
|
|
|
|
+ if queuedCmds, ok := routerQueues[routerID]; ok && len(queuedCmds) > 0 {
|
|
|
|
|
+ log.Printf("Flushing %d queued commands to %s", len(queuedCmds), routerID)
|
|
|
|
|
+ for _, cmd := range queuedCmds {
|
|
|
|
|
+ resultChan := make(chan CommandResult, 1)
|
|
|
|
|
+ cmd.SentAt = time.Now()
|
|
|
|
|
+ pendingCommands[cmd.ID] = resultChan
|
|
|
|
|
+ publishCommand(cmd)
|
|
|
|
|
+ log.Printf("Queued cmd sent: %s", cmd.Command)
|
|
|
|
|
+ // Fire and forget - waiter will handle result
|
|
|
|
|
+ }
|
|
|
|
|
+ delete(routerQueues, routerID)
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
// Message loop
|
|
// Message loop
|
|
|
for {
|
|
for {
|
|
@@ -303,10 +320,6 @@ func handleCommand(w http.ResponseWriter, r *http.Request) {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
router := routers[routerID]
|
|
router := routers[routerID]
|
|
|
- if router == nil || !router.Connected {
|
|
|
|
|
- http.Error(w, "Router not connected", http.StatusServiceUnavailable)
|
|
|
|
|
- return
|
|
|
|
|
- }
|
|
|
|
|
|
|
|
|
|
// ──────────────────────────────────────────────────────────
|
|
// ──────────────────────────────────────────────────────────
|
|
|
// IDEMPOTENCY CHECK - Dedup before publishing!
|
|
// IDEMPOTENCY CHECK - Dedup before publishing!
|
|
@@ -314,7 +327,6 @@ func handleCommand(w http.ResponseWriter, r *http.Request) {
|
|
|
// If client provides an idempotency key, reuse it
|
|
// If client provides an idempotency key, reuse it
|
|
|
xecID := cmd.ID
|
|
xecID := cmd.ID
|
|
|
if xecID == "" {
|
|
if xecID == "" {
|
|
|
- // New command ID acts as idempotency key
|
|
|
|
|
xecID = uuid.New().String()
|
|
xecID = uuid.New().String()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -329,18 +341,32 @@ func handleCommand(w http.ResponseWriter, r *http.Request) {
|
|
|
})
|
|
})
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
- // Old entry, allow retry
|
|
|
|
|
delete(executedCommands, xecID)
|
|
delete(executedCommands, xecID)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-
|
|
|
|
|
- // Create result channel
|
|
|
|
|
- resultChan := make(chan CommandResult, 1)
|
|
|
|
|
|
|
+ // ──────────────────────────────────────────────────────────
|
|
|
|
|
+ // Queue vs Live delivery
|
|
|
|
|
+ // ──────────────────────────────────────────────────────────
|
|
|
cmd.ID = xecID
|
|
cmd.ID = xecID
|
|
|
cmd.RouterID = routerID
|
|
cmd.RouterID = routerID
|
|
|
cmd.SentAt = time.Now()
|
|
cmd.SentAt = time.Now()
|
|
|
|
|
|
|
|
- // Store pending command
|
|
|
|
|
|
|
+ if router == nil || !router.Connected {
|
|
|
|
|
+ // Router offline - QUEUE it!
|
|
|
|
|
+ routerQueues[routerID] = append(routerQueues[routerID], cmd)
|
|
|
|
|
+ log.Printf("Queued command for %s: %s (queue depth: %d)", routerID, cmd.Command, len(routerQueues[routerID]))
|
|
|
|
|
+ json.NewEncoder(w).Encode(map[string]interface{}{
|
|
|
|
|
+ "command_id": cmd.ID,
|
|
|
|
|
+ "status": "queued",
|
|
|
|
|
+ "queued_for": routerID,
|
|
|
|
|
+ "queue_depth": len(routerQueues[routerID]),
|
|
|
|
|
+ "message": "Router offline, command queued",
|
|
|
|
|
+ })
|
|
|
|
|
+ return
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Router online - send directly
|
|
|
|
|
+ resultChan := make(chan CommandResult, 1)
|
|
|
pendingCommands[cmd.ID] = resultChan
|
|
pendingCommands[cmd.ID] = resultChan
|
|
|
|
|
|
|
|
if err := publishCommand(cmd); err != nil {
|
|
if err := publishCommand(cmd); err != nil {
|