rate limitor sketch

rate-limitor.clj

(def state (volatile! {}))
(def c (chan))

(defn add-bucket! [id {:as opts
                       :keys [ttl len]
                       :or {ttl 1000 len 10}}]
  (vswap! state
          assoc
          id
          {:ttl ttl
           :len len
           :queue []}))

(defn add-to-queue! [id x]
  (let [status (atom :ok)]
    (vswap! state
            (fn [s]
              (let [q (get-in s [id :queue])
                    len (get-in s [id :len])]
                (if (< (count q) len)
                  (update-in s [id :queue] conj (assoc x :timestamp (System/currentTimeMillis)))
                  (do (reset! status :full)
                      s)))))
    @status))

(defn clean! []
  (clojure.pprint/pprint @state)
  (println "------------------")
  (vswap! state
          (fn [s]
            (into {}
                  (map (fn [[id {:keys [ttl] :as v}]]
                         [id (update v
                                     :queue
                                     (fn [q]
                                       (vec
                                         (filter (fn [{:keys [timestamp]}]
                                                   (< (- (System/currentTimeMillis) timestamp)
                                                      ttl))
                                                 q))))])
                       s)))))

(comment
  (clojure.pprint/pprint @state)
  (add-bucket! "yop" {:ttl 10000 :len 3})
  (add-to-queue! "yop" {:data 1000}))

(defn start! []
  (vreset! state {})

  ;listener
  (go-loop []
    (let [[type & args] (<! c)
          op (condp = type
               :post add-to-queue!
               :create add-bucket!)
          status (apply op args)]
      (println "status: " status)
      (recur)))

  ;;clean bot
  (future
    (loop []
      (clean!)
      (Thread/sleep 500)
      (recur)))


  ;; create-buckets bot
  (future
    (loop []
      (println "create bot tic")
      (mapv
        (fn [_]
          (async/put! c [:create
                         (keyword (gensym))
                         {:ttl 1000
                          :len (rand-int 8)}]))
        (range (rand-int 4)))
      (Thread/sleep 5000)
      (recur)))


  ;;post-bot
  (future
    (loop []
      (println "post bot tic")
      (mapv
        (fn [_]
          (when-let [id (rand-nth (keys @state))]
            (async/put! c [:post id {:data (rand-int 200)}])))
        (range (rand-int 20)))
      (Thread/sleep (rand-int 3000))
      (recur))))

(comment (start!))