async-timeline.clj

(defn fetch-user
  [id]
  (fetch-item id))

(defn fetch-post
  [id]
  (fetch-item id))

(defn timeline-ids
  [username]
  (fetch-collection (str "@" username)))

(defn identibale-user
  [id]
  (async/map (partial vector id) [(fetch-user id)]))

(defn inject-score
  [scores [id info]]
  [id (assoc info :score (scores id))])

(defn inject-user
  [users post]
  (let [id (get-in post [:owner :id])]
    (assoc post :owner (users id))))

(defn timeline-posts
  [username]
  (go
    (let [ids (<! (timeline-ids username))
          posts-chs (map fetch-posts (take 20 ids))
          posts (<! (async/map vector post-chs))
          user-ids (set (map #(get-in % [:owner :id]) posts))
          users-chs (map identibale-user user-ids)
          scores-ch (all-user-scores user-ids)
          user-infos (into {} (<! (async/map vector users-chs)))
          scores (<! scores-ch)
          users (->> user-infos
                     (map (partial inject-score scores))
                     (into {}))]
      (map (partial inject-user users) posts)))

damn-timeline.clj

(defn fetch-user [id]
  (let [user (fetch-item id)
        score (fetch-score id)]
    (assoc user :score score)))

(defn fetch-post [id]
  (let [post (fetch-item id)
        user (fetch-user (get-in post [:owner :id]))]
    (assoc post :owner user)))

(defn timeline-ids [username]
  (fetch-collection (str "@" username)))

(defn timeline-posts [username]
  (->> username
       timeline-ids
       (take 20)
       (map fetch-post)))

friends-of.clj

user=> (require '[muse.core :refer :all])
nil
user=> (require '[clojure.core.async :refer [go <!!]])
nil
user=> (defrecord FriendsOf [id]
  #_=>   DataSource
  #_=>   (fetch [_] (go (range id))))
user.FriendsOf
user=> (FriendsOf. 10)
#user.FriendsOf{:id 10}

user=> (run! (FriendsOf. 10))
#<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@5204116f>
user=> (<!! (run! (FriendsOf. 10)))
(0 1 2 3 4 5 6 7 8 9)
user=> (run!! (FriendsOf. 10))
(0 1 2 3 4 5 6 7 8 9)

user=> (fmap count (FriendsOf. 10))
#<MuseMap (clojure.core$count@34323247 user.FriendsOf[10])>
user=> (run!! (fmap count (FriendsOf. 10)))
10
user=> (fmap inc (fmap count (FriendsOf. 3)))
#<MuseMap (clojure.core$comp$fn__4192@235770e0 user.FriendsOf[3])>
user=> (run!! (fmap inc (fmap count (FriendsOf. 3))))
4

user=> (defrecord PostsCount [id]
  #_=>   DataSource
  #_=>   (fetch [_] (go (inc id))))
user.PostsCount
user=> (flat-map #(PostsCount. %) (PostsCount. 10))
#<MuseFlatMap (user$eval10509$fn__10510@6523b02f user.PostsCount[10])>
user=> (run!! (flat-map #(PostsCount. %) (PostsCount. 10)))
12
user=> (traverse #(PostsCount. %) (FriendsOf. 10))
#<MuseFlatMap (muse.core$traverse$fn__10255@3dc2864 user.FriendsOf[10])>
user=> (run!! (traverse #(PostsCount. %) (FriendsOf. 10)))
[1 2 3 4 5 6 7 8 9 10]

user=> (run!! (->> (FriendsOf. 5)
  #_=>             (traverse #(FriendsOf. %))
  #_=>             (fmap (partial apply concat))
  #_=>             (fmap set)))
#{0 1 3 2}

muse-repl.clj

user=> (require '[muse.core :refer :all] :reload)
nil
user=> (require '[clojure.core.async :refer [go]])
nil
user=> (defrecord Range [id]
  #_=>   DataSource
  #_=>   (fetch [_] (go (range id))))
user.Range
user=> (Range. 10)
#user.Range{:id 10}

user=> (run! (Range. 10))
#<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@63a01449>
user=> (<!! (run! (Range. 10)))
(0 1 2 3 4 5 6 7 8 9)
user=> (run!! (Range. 5))
(0 1 2 3 4)

user=> (fmap count (Range. 10))
#<MuseMap (clojure.core$count@3d804ede user.Range[10])>
user=> (run!! (fmap count (Range. 10)))
10
user=> (fmap inc (fmap count (Range. 3)))
#<MuseMap (clojure.core$comp$fn__4192@58dc9797 user.Range[3])>
user=> (run!! (fmap inc (fmap count (Range. 3))))
4

user=> (defrecord Inc [id]
  #_=>   DataSource
  #_=>   (fetch [_] (go (inc id))))
user=> (flat-map ->Inc (->Inc 3))
#<MuseFlatMap (user$eval10466$__GT_Inc__10498@411c0aeb user.Inc[3])>
user=> (run!! (flat-map ->Inc (->Inc 3)))
5
user=> (run!! (flat-map ->Inc (fmap count (Range. 4))))
5
user=> (traverse ->Inc (Range. 3))
#<MuseFlatMap (muse.core$traverse$fn__10255@7ed127e0 user.Range[3])>
user=> (run!! (traverse ->Inc (Range. 3)))
[1 2 3]

muse-sql.clj

;;
;; Muse library - https://goo.gl/BjWaZa
;;

(require '[clojure.string :as s])
(require '[clojure.core.async :as async :refer [<!]])
(require '[muse.core :refer :all])
(require '[postgres.async :refer :all])

(defrecord Posts [limit]
  DataSource
  (fetch [_]
    (async/map :rows [(execute! db ["select id, user, title, text from posts limit $1" limit])])))

(defrecord User [id]
  DataSource
  (fetch [_]
    (async/map :rows [(execute! db ["select id, name from users where id = $1"])]))
  
  BatchedSource
  (fetch-multi [_ users]
    (let [all-ids (cons id (map :id users))
          query (str "select is, name from users where id IN (" (s/join "," all-ids) ")")]
      (go
       (let [{:keys [rows]} (<! (execute! db [query]))]
         (into {} (map (fn [{:keys [id] :as row}] [id row]) rows)))))))

(defn attach-author [{:keys [user] :as post}]
  (fmap #(assoc post :user %) (User. user)))

(def fetch-posts [limit]
  (traverse attach-author (Posts. limit)))

;; will execute 2 SQL queries instead of 11
(run!! (fetch-posts 10))

muse-timeline.clj

(defn fetch-user [id]
  (fmap #(assoc %1 :score %2) (User. id) (UserScore. id)))

(defn attach-author [{:keys [owner] :as post}]
  (fmap #(assoc post :owner %) (fetch-user (:id owner))))

(defn fetch-post [id]
  (flat-map attach-author (Post. id)))

(defn timeline-ids [username]
  (Timeline. username))

(defn timeline-posts [username]
  (->> username
       timeline-ids
       (fmap #(take 20 %))
       (traverse fetch-post)))