"Don't know how to create ISeq from: clojure.core.async.impl.channels.ManyToManyChannel" when using transducers with async.channel

I'm learning core.async from book "Reactive programming in clojure"

One of the sample program is below -


(ns core-async-intro.xforms
  (:require [clojure.core.async :refer [map< filter< chan go <! >! close!]]))

(def result (chan 10))

(def transformed
  (->> result
       (map< inc) ;; creates a new channel
       (filter< even?) ;; creates a new channel
       (into [])))

(defn -main [& args]
  (go
    (prn "result is " (<! transformed)))

  (go
    (doseq [n (range 10)]
      (>! result n))
    (close! result)))

But when I run this program I get below error -

{:clojure.main/message

"Execution error (IllegalArgumentException) at user/eval140$fn (form-init12172038470509011992.clj:1).\nDon't know how to create ISeq from: clojure.core.async.impl.channels.ManyToManyChannel\n",

:clojure.main/triage

{:clojure.error/class java.lang.IllegalArgumentException,

:clojure.error/line 1

:clojure.error/cause

"Don't know how to create ISeq from: clojure.core.async.impl.channels.ManyToManyChannel",

:clojure.error/symbol user/eval140$fn,

:clojure.error/source "form-init12172038470509011992.clj",

:clojure.error/phase :execution},

:clojure.main/trace

{:via

[{:type clojure.lang.Compiler$CompilerException,
:message "Syntax error macroexpanding at (xforms.clj:10:8).",
 
:data
 
{:clojure.error/phase :execution,

:clojure.error/line 10,

:clojure.error/column 8,

:clojure.error/source "xforms.clj"},

:at [clojure.lang.Compiler$InvokeExpr eval "Compiler.java" 3719]}

{:type java.lang.IllegalArgumentException,

:message

"Don't know how to create ISeq from: clojure.core.async.impl.channels.ManyToManyChannel",

:at [clojure.lang.RT seqFrom "RT.java" 557]}],

clojure version is - 1.12.0 and clojure.core.async - 1.6.681

How can I fix this error?

Solution

You are using clojure.core/into as the last step for transformed. That works with sequences, not core.async channels. So you need to switch that to the async variant.

In general for async operations I would recommend to a style that makes that clear and do not use :refer so much.

(ns core-async-intro.xforms
  (:require [clojure.core.async :as async :refer [go <! >!]]))

(def result (async/chan 10))

(def transformed
  (->> result
       (async/map< inc) ;; creates a new channel
       (async/filter< even?) ;; creates a new channel
       (async/into [])))

(defn -main [& args]
  (go
    (prn "result is " (<! transformed)))

  (go
    (doseq [n (range 10)]
      (>! result n))
    (async/close! result)))