Start and Stop core.async Interval

Question

consider the following core-async code. It prints the string "tick" every 2 seconds:

(go-loop []
  (<! (timeout 2000))
  (prn "tick")
  (recur))

I'm looking for a possibility to start and stop the interval from outside via functions.

The only thing that came to my mind was this:

(def running (atom false))

(defn start []
  (reset! running true)
  (go (loop []
         (<! (timeout 2000))
         (prn "tick")
         (when @running (recur)))))

(defn stop []
  (reset! running false))

Is this the way to go? Or would you do something else?

Answer 1

Here's what I would do:

(require '[clojure.core.async :as a])

(defn interval [f msecs]
  (let [timing (a/chan)
        kickoff
        #(a/go
           (a/<! (a/timeout msecs))
           (a/>! timing true))]
    (a/go-loop []
      (when (a/<! timing)
        (a/go (f))
        (kickoff)
        (recur)))
    (kickoff)
    #(a/close! timing)))

Example:

(let [i (interval #(prn "tick") 2000)]
  (Thread/sleep 7000)
  (i))
;; "tick"
;; "tick"
;; "tick"
;;=> nil

This way, all your state is local and is handled via channels and go blocks, rather than using a global atom. Since the work done in each iteration of the loop is more or less constant, your actual interval time should be fairly close to the number of msecs you pass in.

Or if you want to guarantee that the different calls to f are executed in chronological order, you could do something like this instead:

(require '[clojure.core.async :as a])

(defn interval [f msecs]
  (let [action (a/chan (a/dropping-buffer 1))
        timing (a/chan)
        kickoff
        #(a/go
           (a/<! (a/timeout msecs))
           (a/>! timing true))]
    (a/go-loop []
      (when (a/<! action)
        (f)
        (recur)))
    (a/go-loop []
      (if (a/<! timing)
        (do
          (a/>! action true)
          (kickoff)
          (recur))
        (a/close! action)))
    (kickoff)
    #(a/close! timing)))

(The usage of this function is the same as that of my earlier interval function.)

Here all the calls to f are in the same loop. I'm using a dropping buffer, so if a call to f takes longer than msecs, future calls may start getting dropped to keep up.

Answer 2

Here is another way, you could go

(defn start []
 (let [ctl (chan 0 )]
   (go-loop []
     (alt!
       (timeout 2000) (do
                     (prn "tick")
                     (recur))
        ctl (prn "Exit ")))
   ctl))


(let [w (start) ]
  (Thread/sleep 7000)
  (>!! w "stop"))

;;=>     "tick"
         "tick"
         "Exit"     

You don't need to define state in namespace. Start function is returning chan to do stop in future.