keyword union path breaks on async parsers

[wilkerlucio]

No description provided.

[yenda]

I had this issue and used the following code to fix it in my project. Is it fixed in the latest version of pathon already?

(defn join-seq-parallel [{:keys  [query]
                          ::p/keys [entity-path-cache parent-query]
                          :as    env} coll]
  (let [query' (if (nat-int? query) parent-query query)]
    (if (seq coll)
      (go-promise
       (pt/tracing env {::pt/event ::p/parallel-sequence-loop
                        ::pt/style {:fill    "#e0e3a4"
                                    :opacity "0.8"}}
                   (let [ast            (eql/query->ast query')
                         check-ast-opt? (every? #(not (:children %)) (:children ast))
                         join-item      (fn join-item [env entity]
                                          (or (and
                                               check-ast-opt?
                                               (reduce
                                                (fn [ent {:keys [key params]}]
                                                  (if-let [[_ v] (find entity key)]
                                                    (assoc ent (get params :pathom/as key) v)
                                                    (reduced nil)))
                                                {}
                                                (:children ast)))
                                              (p/join entity env)))
                         env            (assoc env ::p/processing-sequence coll)
                         [head & tail] coll
                         first-res      (<?maybe (join-item (update env ::path conj 0) head))
                         from-chan      (async/chan 10)
                         out-chan       (async/chan 10)]
                     (async/onto-chan from-chan (map vector tail (range)))
                     (async/pipeline-async 10
                                           out-chan
                                           (fn join-seq-pipeline [[ent i] res-ch]
                                             (go
                                               (let [{::p/keys [path] :as env'} (update env ::path conj (inc i))
                                                     ent (merge (get @entity-path-cache path {}) ent)
                                                     res (<!maybe (join-item env' ent))]
                                                 (>! res-ch res)
                                                 (async/close! res-ch))))
                                           from-chan)
                     (<! (async/into [first-res] out-chan)))))
      [])))

(def user-notifications
  {:user/notifications
   (fn [{:keys [authed-user-id redis-connection ast] :as env}]
     (go-catch
      (let [{:keys [limit offset]} (:params ast)
            notifications (<! (models.notifications/get-user-notifications redis-connection
                                                                           authed-user-id
                                                                           offset limit))]
        (if (seq notifications)
          (<! (join-seq-parallel (assoc env
                                        ::p/union-path
                                        (fn [env]
                                          (let [e (p/entity env)]
                                            (:notification/type e))))
                                 notifications))
          []))))})