Clojure - используйте канал core.async с Yada/Aleph

Question

Я пытаюсь использовать Clojure manifold библиотека, и чтобы понять это, мне нужно преобразовать core.async канал в коллекторный поток.

Я хотел бы создать эквивалент следующего, используя core.async канал:

 (require '[manifold.stream :as s])
 (s/periodically 100 #(str " ok "))

  ;; Here is what I tried, it fails with an error 500
  (let [ch (chan)]
      (go-loop []
      (>! ch " ok ")
      (<! (timeout 100))
      (recur))
    (s/->source ch))

Я пытаюсь накормить core.async канал в yada, Первый пример кода, используя manifold.stream/periodic работает, а не другие, использующие core.async, Я примеряла яду 1.0.0 а также 1.1.0-SNAPSHOT,

С помощью manifold.stream/periodic работает:

(def get-stream
  (yada (fn [ctx]
          (-> (:response ctx)
              (assoc :status 202)
              (assoc :body (s/periodically 1000 #(str (System/currentTimeMillis) " ")))))
        {:representations [{:media-type "application/json"
                            :charset    "UTF-8"}
                           {:media-type "application/edn"
                            :charset    "UTF-8"}]}))

С помощью manifold.stream/->source возвращает ошибку 500:

(def get-stream
  (yada (fn [ctx]
          (-> (:response ctx)
              (assoc :status 202)
              ;; Similar to this : https://github.com/juxt/yada/blob/94f3ee93de155a8513b27e0508608691ed556a55/dev/src/yada/dev/async.clj
              (assoc :body (let [ch (chan)]
                             (go-loop []
                               (>! ch " ok ")
                               (<! (timeout 100))
                               (recur))
                             (s/->source ch)))))
        {:representations [{:media-type "application/json"
                            :charset    "UTF-8"}
                           {:media-type "application/edn"
                            :charset    "UTF-8"}]}))

;; Error on the page :

;; 500: Unknown
;; Error on GET

;; #error {
;;         :cause "No implementation of method: :to-body of protocol: #'yada.body/MessageBody found for class: manifold.stream.async.CoreAsyncSource"
;;         :via
;;         [{:type clojure.lang.ExceptionInfo
;;           :message "Error on GET"
;;           :data {:response #yada.response.Response{:representation {:media-type #yada.media-type[application/json;q=1.0], :charset #yada.charset.CharsetMap{:alias "UTF-8", :quality 1.0}}, :vary #{:media-type}}, :resource #function[backend.routes.examples.ts/fn--57734]}
;;                                                                                                  :at [clojure.core$ex_info invoke "core.clj" 4593]}
;;                                                                     {:type java.lang.IllegalArgumentException
;;                                                                      :message "No implementation of method: :to-body of protocol: #'yada.body/MessageBody found for class: manifold.stream.async.CoreAsyncSource"
;;                                                                      :at [clojure.core$_cache_protocol_fn invoke "core_deftype.clj" 554]}]
;;                                                    :trace

Третья попытка, с core.async канал (другая ошибка 500):

(def get-stream
  (yada (fn [ctx]
          (-> (:response ctx)
          (assoc :status 202)
          (assoc :body (chan)))
        {:representations [{:media-type "application/json"
                            :charset    "UTF-8"}
                           {:media-type "application/edn"
                            :charset    "UTF-8"}]}))

;; Error on the page :

;; 500: Unknown

;; Error on GET

;; #error {
;;         :cause "No implementation of method: :to-body of protocol: #'yada.body/MessageBody found for class: clojure.core.async.impl.channels.ManyToManyChannel"
;;         :via
;;         [{:type clojure.lang.ExceptionInfo
;;           :message "Error on GET"
;;           :data {:response #yada.response.Response{:representation {:media-type #yada.media-type[application/json;q=1.0], :charset #yada.charset.CharsetMap{:alias "UTF-8", :quality 1.0}}, :vary #{:media-type}}, :resource #function[backend.routes.api.subscribe/subscribe$fn--64130]}
;;                                                                                                  :at [clojure.core$ex_info invoke "core.clj" 4593]}
;;                                                                     {:type java.lang.IllegalArgumentException
;;                                                                      :message "No implementation of method: :to-body of protocol: #'yada.body/MessageBody found for class: clojure.core.async.impl.channels.ManyToManyChannel"
;;                                                                      :at [clojure.core$_cache_protocol_fn invoke "core_deftype.clj" 554]}]
;;                                                    :trace

Answer 1

Ключевая ошибка заключается в следующем:

"No implementation of method: :to-body of protocol: 
 #'yada.body/MessageBody     
 found for class: manifold.stream.async.CoreAsyncSource"

Это показывает, что используемая вами версия yada пытается привести к типу тела, которого она не понимает. Более поздние версии yada более разборчивы в том, что вы можете отправлять через веб-сервер, и разрешают все, что он не знает, как преобразовать.