Port of rx video example from Groovy to Clojure

video_example.clj

(ns rx.examples.clojure.video-example
  (:import [rx Observable Observer Subscription]))

(defn video-list
  [position]
  {:position position
   :name (str "ListName-" position) })

(defn ^Observable video-list->videos
  [{:keys [position] :as video-list}]
  (Observable/create (fn [^Observer observer]
                       (dotimes [i 50]
                         (.onNext observer (+ (* position 1000) i)))
                       (.onCompleted observer))))

(comment (.subscribe (video-list->videos (video-list 2)) println))

(defn ^Observable video->metadata
  [video-id]
  (Observable/create (fn [^Observer observer]
                       (.onNext observer {:title (str "video-" video-id "-title")
                                          :actors ["actor1" "actor2"]
                                          :duration 5428 })
                       (.onCompleted observer)
                       nil)))

(comment (.subscribe (video->metadata 10) println))

(defn ^Observable video->bookmark
  [video-id user-id]
  (Observable/create (fn [^Observer observer]
                       (future
                         (Thread/sleep 4)
                         (.onNext observer (if (> (rand-int 6) 1) 0 (rand-int 4000)))
                         (.onCompleted observer))
                       nil)))

(comment (.subscribe (video->bookmark 112345 99999) println))

(defn ^Observable video->rating
  [video-id user-id]
  (Observable/create (fn [^Observer observer]
                       (future
                         (Thread/sleep 10)
                         (.onNext observer {:video-id              video-id
                                            :user-id               user-id
                                            :predicted-star-rating (rand-int 5)
                                            :average-star-rating   (rand-int 5)
                                            :actual-star-rating    (rand-int 5) }) )
                       nil)))

(comment (.subscribe (video->rating 234345 8888) println))

(defn ^Observable get-list-of-lists
  "
  Retrieve a list of lists of videos (grid).

  Observable<VideoList> is the "push" equivalent to List<VideoList>
  "
  [user-id]
  (Observable/create (fn [^Observer observer]
                       (future
                         (Thread/sleep 180)
                         (dotimes [i 15]
                           (.onNext observer (video-list i)))
                         (.onCompleted observer))
                       nil)))

(comment (.subscribe (get-list-of-lists 7777) println))

(defn ^Observable get-video-grid-for-display
  "
  Demonstrate how Rx is used to compose Observables together such as
  how a web service would to generate a JSON response.

  The simulated methods for the metadata represent different services
  that are often backed by network calls.

  This will return a sequence of maps like this:

   {:id 1000, :title video-1000-title, :length 5428, :bookmark 0,
    :rating {:actual 4 :average 3 :predicted 0}}
  "
  [user-id]
  (-> (get-list-of-lists user-id)
    (.mapMany (fn [list]
                ; for each VideoList we want to fetch the videos
                (-> (video-list->videos list)
                  (.take 10) ; we only want the first 10 of each list
                  (.mapMany (fn [video]
                              ; for each video we want to fetch metadata
                              (let [m (-> (video->metadata video)
                                          (.map (fn [md]
                                                  ; transform to the data and format we want
                                                  {:title  (:title md)
                                                  :length (:duration md) })))
                                    b (-> (video->bookmark video user-id)
                                          (.map (fn [position]
                                                  {:bookmark position})))
                                    r (-> (video->rating video user-id)
                                          (.map (fn [rating]
                                                  {:rating {:actual    (:actual-star-rating rating)
                                                            :average   (:average-star-rating rating)
                                                            :predicted (:predicted-star-rating rating) }})))]
                                ; join these together into a single, merged map for each video
                                (Observable/zip m b r (fn [m b r]
                                                        (merge {:id video} m b r)))))))))))

(defn -main
  [& args]
  ; this will print the dictionary for each video and is a good representation of
  ; how progressive rendering could work
  (println "---- sequence of video dictionaries ----")
  (comment (-> (get-video-grid-for-display 1)
    (.subscribe #(locking *ns* (println %))
                #(locking *ns* (println "Error: " %)))))

  ; onNext will be called once with a list and demonstrates how a sequence can be combined
  ; for document style responses (most webservices)
  (-> (get-video-grid-for-display 1)
    .toList
    (.subscribe #(locking *ns* (println "\n ---- single list of video dictionaries ----\n" %))
                #(locking *ns* (println "Error: " %)))))

(comment (-main))