Using receive In A Helper Function

[pwhittin]

Within event_test.clj you'll notice that do-receive is a macro. I had to make it a macro because it did not work as a function. It compiles, but when the tests are executed I get the following error associated with the call to receive!:

java.lang.AssertionError: Assert failed: alts! used not in (go ...) block

The following code is the implementation I tested of do-receive as a function:

(defn do-receive
  ([done-name] (do-receive done-name true 0 false))
  ([done-name unique] (do-receive done-name false unique true))
  ([done-name done-is unique unique-is]
   (op/receive!
     [unique :ok]
     (is unique-is "received [ref :ok] message")

     [:done done-name]
     (is done-is "to-go timeout reached")

     :else
     (is false "unknown message")

     (after 2000 (is false "timeout occurred")))))

The odd thing is that the following similar code executes perfectly in a newly started lein repl:

(require '[otplike.process :as process :refer [! pid->str self]])
(require '[clojure.core.async :as async])

(defn my-receive
  [msg1 msg2 timeout self]
  (process/receive!
    msg (printf "[%s] %s %s%n" self msg1 msg)
    (after timeout
           (printf "[%s] %s %s%n" self msg2 timeout))))

(process/proc-defn
  await-message
  [timeout]
  (my-receive "receive" "timeout" timeout (process/self)))

(defn run []
  (let [pid0 (process/spawn await-message [1000])]
    (println "[main] Started [" pid0 "]")
    (let [pid1 (process/spawn await-message [(async/timeout 1000)])]
      (println "[main] Started [" pid1 "]")
      (let [pid2 (process/spawn await-message [:infinity])]
        (println "[main] Started [" pid2 "]")
        (println "[main] Waiting for 2 seconds")
        (Thread/sleep 2000)
        (println "[main] Sending :stop to [" pid2 "]")
        (! pid2 :stop)
        (println "[main] Waiting for 100mS")
        (Thread/sleep 100)
        (println "[main] Done")))))

(run)

[sskorokhodov]

Thank you for the question. It's not an obvious thing. Maybe we should extend docs to explain this.

First. Try to change await-message like this:

(process/proc-defn watcher []
  (process/receive!
    msg (do (printf "%s%n" msg) (flush))))

(process/proc-defn
  await-message
  [timeout]
  (process/spawn-opt watcher {:link true :flags {:trap-exit true}})
  (my-receive "receive" "timeout" timeout (process/self)))

You'll see that the problem is still there.

What happened before the changes: because no process watches await-message and tracing is off, the process crashes silently.

Second. We try to keep the semantics of ! and !! at the end of macro's name the same as in core.async. Therefore when for some reason you want to receive messages not in go-block and blocking is not an issue, receive!! can be used instead of receive!.

[pwhittin]

As you suggested, using "receive!!" allowed me to make do-receive just a regular function:

(defn do-receive
  ([done-name] (do-receive done-name true 0 false))
  ([done-name unique] (do-receive done-name false unique true))
  ([done-name done-is unique unique-is]
   (op/receive!!                  ;; <<<<<===== Here is the change
     [unique :ok]
     (is unique-is "received [ref :ok] message")

     [:done done-name]
     (is done-is "to-go timeout reached")

     :else
     (is false "unknown message")

     (after 2000 (is false "timeout occurred")))))