durable subscriptions, tweak rescues & headcodes & statsd

trackernet.lisp

@@ -4471,6 +4471,7 @@ Return four values: a set of nodes, edges, weights, and a new->old edge mapping.
     (return-from stomp-td-message-handler))
   (unless (string= (cdr (assoc :area--id body)) "Q0")
     (return-from stomp-td-message-handler))
+  (statsd-inc "intertube.xr-stomp-message")
   (when (and (cdr (assoc :descr body))
              (position #\* (cdr (assoc :descr body))))
     ;; FIXME(eta): hotpatch to prevent asterisks from getting in
@@ -4556,7 +4557,8 @@ Return four values: a set of nodes, edges, weights, and a new->old edge mapping.
         (log:info "interpose in berth ~A links ~A to ~A, rescuing"
                   berth rescue-headcode headcode)
         (rescue-train (format nil "X-train-~A" rescue-headcode)
-                      (format nil "X-train-~A" headcode))))))
+                      (format nil "X-train-~A" headcode))
+        (red:del (format nil "X-train-~A" rescue-headcode))))))
 
 (defvar *current-stomp-td-backoff* 1)
 
@@ -4594,6 +4596,15 @@ Return four values: a set of nodes, edges, weights, and a new->old edge mapping.
     (stomp:stop *stomp-td*)
     (setf *stomp-td* nil)))
 
+;; Start a durable subscription, using the machine hostname as the
+;; subscription name
+(defmethod stomp::send :around (conn frame)
+  (when (string= (stomp:frame-name frame) "CONNECT")
+    (stomp:set-header frame "client-id" (car *nrod-data-key*)))
+  (when (string= (stomp:frame-name frame) "SUBSCRIBE")
+    (stomp:set-header frame "activemq.subscriptionName" (machine-instance)))
+  (call-next-method))
+
 (defun stomp-td-connect ()
   "Connect to the National Rail TD feeds over STOMP."
   (unless *nrod-data-key*

wobsite.lisp

@@ -2037,7 +2037,6 @@
       for lt in train-list
       do (alexandria:when-let
              ((wtt (trackernet::trackernet-id lt)))
-           (setf wtt (subseq wtt 3))
            (setf (gethash wtt wtts)
                  (cons
                   lt