   (if (empty? downstream)
     []
     [[:message-envelope
-      (rand-nth (seq downstream))
+      (rand-nth* (seq downstream))
       :forwarded-subscription
       subscription
       (get-envelope-id)]]))


    ;; Forward subscription to :downstream.
    (let [downstream (seq downstream)
          downstream+ (reduce (fn [x _] (conj x (rand-nth downstream)))
                              downstream
                              (range (:connection-redundancy config)))]
      (map

  "TODO: replace with UUIDs"
  []
  (str (swap! envelope-id-counter inc)))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Custom randomization (for predictable testing)
(def ^:dynamic *rand* rand)

(defn- ^Integer rand-int* [n]
  (int (*rand* n)))

(defn- rand-nth* [coll]
  (nth coll (rand-int* (count coll))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Schema definitions
