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. Я пробовал на yada 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 более разрешительны в отношении того, что вы можете отправить на веб-сервер, и разрешите что-либо через это, он не знает, как преобразовать.