Merge pull request #834 from daveray/update-clojure-for-0.17

Update clojure for 0.17

Author: benjchristensen

language-adaptors/rxjava-clojure/README.md

@@ -46,6 +46,19 @@ The `rx/action` macro is identical to `rx/fn` except that the object returned im
                 (rx/action [] (println "Sequence complete"))))
 ```
 
+## Using Observable/create
+As of 0.17, `rx.Observable/create` takes an implementation of `rx.Observable$OnSubscribe` which is basically an alias for `rx.util.functions.Action1` that takes an `rx.Subscriber` as its argument. Thus, you can just use `rx/action` when creating new observables:
+
+```clojure
+; A simple observable that emits 0..9 taking unsubscribe into account
+(Observable/create (rx/action [^rx.Subscriber s]
+                     (loop [i 0]
+                       (when (and (< i 10) (.isUnsubscribed s))
+                         (.onNext s i)
+                         (recur (inc i))))
+                     (.onCompleted s)))
+```
+
 # Gotchas
 Here are a few things to keep in mind when using this interop:
 

language-adaptors/rxjava-clojure/build.gradle

@@ -50,3 +50,26 @@ jar {
         instruction 'Fragment-Host', 'com.netflix.rxjava.core'
     }
 }
+
+
+////////////////////////////////////////////////////////////////////////////////
+// Define a task that runs an nrepl server. The port is given with the nreplPort
+// property:
+//    gradlew nrepl -PnreplPort=9999
+// or put the property in ~/.gradle/gradle.properties
+
+def nreplPortValue = (project.hasProperty('nreplPort') && !project.nreplPort.isEmpty()) ? project.nreplPort : '9999'
+configurations { nrepl }
+dependencies { nrepl 'org.clojure:tools.nrepl:0.2.2' }
+task nrepl(type: JavaExec) {
+    classpath configurations.nrepl,
+              project.sourceSets.main.clojure.srcDirs,
+              project.sourceSets.test.clojure.srcDirs,
+              sourceSets.main.runtimeClasspath,
+              sourceSets.test.runtimeClasspath
+
+    main = "clojure.main"
+    args '--eval', "(ns gradle-nrepl (:require [clojure.tools.nrepl.server :refer (start-server stop-server)]))",
+         '--eval', "(println \"Starting nrepl server on port $nreplPortValue\")",
+         '--eval', "(def server (start-server :port $nreplPortValue))"
+}

language-adaptors/rxjava-clojure/src/examples/clojure/rx/lang/clojure/examples/http_examples.clj

@@ -1,12 +1,12 @@
 ;
 ; Copyright 2013 Netflix, Inc.
-;  
+;
 ; Licensed under the Apache License, Version 2.0 (the "License");
 ; you may not use this file except in compliance with the License.
 ; You may obtain a copy of the License at
 ;
 ; http://www.apache.org/licenses/LICENSE-2.0
-; 
+;
 ; Unless required by applicable law or agreed to in writing, software
 ; distributed under the License is distributed on an "AS IS" BASIS,
 ; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -27,14 +27,14 @@
 
    return Observable<String> of HTML"
   (Observable/create
-    (rx/fn [observer]
+    (rx/action [observer]
       (let [f (future
                 (doseq [articleName wikipediaArticleNames]
                   (-> observer (.onNext (http/get (str "http://en.wikipedia.org/wiki/" articleName)))))
                 ; after sending response to onnext we complete the sequence
                 (-> observer .onCompleted))]
         ; a subscription that cancels the future if unsubscribed
-        (Subscriptions/create (rx/action [] (future-cancel f)))))))
+        (.add observer (Subscriptions/create (rx/action [] (future-cancel f))))))))
 
 ; To see output
 (comment
@@ -52,7 +52,7 @@
 
    return Observable<String> of HTML"
   (Observable/create
-    (rx/fn [observer]
+    (rx/action [observer]
       (let [f (future
                 (try
                   (doseq [articleName wikipediaArticleNames]
@@ -62,7 +62,7 @@
                 ; after sending response to onNext we complete the sequence
                 (-> observer .onCompleted))]
         ; a subscription that cancels the future if unsubscribed
-        (Subscriptions/create (rx/action [] (future-cancel f)))))))
+        (.add observer (Subscriptions/create (rx/action [] (future-cancel f))))))))
 
 ; To see output
 (comment

language-adaptors/rxjava-clojure/src/examples/clojure/rx/lang/clojure/examples/rx_examples.clj

@@ -1,12 +1,12 @@
 ;
 ; Copyright 2013 Netflix, Inc.
-;  
+;
 ; Licensed under the Apache License, Version 2.0 (the "License");
 ; you may not use this file except in compliance with the License.
 ; You may obtain a copy of the License at
 ;
 ; http://www.apache.org/licenses/LICENSE-2.0
-; 
+;
 ; Unless required by applicable law or agreed to in writing, software
 ; distributed under the License is distributed on an "AS IS" BASIS,
 ; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -58,53 +58,37 @@
 ; Custom Observable
 ; --------------------------------------------------
 
-(defn customObservableBlocking []
-  "This example shows a custom Observable that blocks
-   when subscribed to (does not spawn an extra thread).
+(defn customObservable
+  "This example shows a custom Observable. Note the
+  .isUnsubscribed check so that it can be stopped early.
 
   returns Observable<String>"
+  []
   (Observable/create
-    (rx/fn [observer]
-      (doseq [x (range 50)] (-> observer (.onNext (str "value_" x))))
+    (rx/action [^rx.Subscriber s]
+      (loop [x (range 50)]
+        (when (and (not (.isUnsubscribed s)) x)
+          ; TODO
+          (println "HERE " (.isUnsubscribed s) (first x))
+          (-> s (.onNext (str "value_" (first x))))
+          (recur (next x))))
       ; after sending all values we complete the sequence
-      (-> observer .onCompleted)
-      ; return a NoOpSubsription since this blocks and thus
-      ; can't be unsubscribed from
-      (Subscriptions/empty))))
+      (-> s .onCompleted))))
 
 ; To see output
 (comment
-  (.subscribe (customObservableBlocking) (rx/action* println)))
-
-(defn customObservableNonBlocking []
-  "This example shows a custom Observable that does not block
-  when subscribed to as it spawns a separate thread.
-
-  returns Observable<String>"
-  (Observable/create
-    (rx/fn [observer]
-      (let [f (future
-                (doseq [x (range 50)]
-                  (-> observer (.onNext (str "anotherValue_" x))))
-                ; after sending all values we complete the sequence
-                (-> observer .onCompleted))]
-        ; return a subscription that cancels the future
-        (Subscriptions/create (rx/action [] (future-cancel f)))))))
-
-; To see output
-(comment
-  (.subscribe (customObservableNonBlocking) (rx/action* println)))
-
+  (.subscribe (customObservable) (rx/action* println)))
 
 ; --------------------------------------------------
 ; Composition - Simple
 ; --------------------------------------------------
 
-(defn simpleComposition []
-  "Asynchronously calls 'customObservableNonBlocking' and defines
+(defn simpleComposition
+  "Calls 'customObservable' and defines
    a chain of operators to apply to the callback sequence."
+  []
   (->
-    (customObservableNonBlocking)
+    (customObservable)
     (.skip 10)
     (.take 5)
     (.map (rx/fn [v] (str v "_transformed")))
@@ -119,66 +103,73 @@
 ; Composition - Multiple async calls combined
 ; --------------------------------------------------
 
-(defn getUser [userId]
+(defn getUser
   "Asynchronously fetch user data
 
   return Observable<Map>"
+  [userId]
   (Observable/create
-    (rx/fn [observer]
+    (rx/action [^rx.Subscriber s]
       (let [f (future
                 (try
                   ; simulate fetching user data via network service call with latency
                   (Thread/sleep 60)
-                  (-> observer (.onNext {:user-id userId
-                                         :name "Sam Harris"
-                                         :preferred-language (if (= 0 (rand-int 2)) "en-us" "es-us") }))
-                  (-> observer .onCompleted)
-                  (catch Exception e (-> observer (.onError e))))) ]
+                  (-> s (.onNext {:user-id userId
+                                  :name "Sam Harris"
+                                  :preferred-language (if (= 0 (rand-int 2)) "en-us" "es-us") }))
+                  (-> s .onCompleted)
+                  (catch Exception e
+                    (-> s (.onError e))))) ]
         ; a subscription that cancels the future if unsubscribed
-        (Subscriptions/create (rx/action [] (future-cancel f)))))))
+        (.add s (Subscriptions/create (rx/action [] (future-cancel f))))))))
 
-(defn getVideoBookmark [userId, videoId]
+(defn getVideoBookmark
   "Asynchronously fetch bookmark for video
 
   return Observable<Integer>"
+  [userId, videoId]
   (Observable/create
-    (rx/fn [observer]
+    (rx/action [^rx.Subscriber s]
       (let [f (future
                 (try
                   ; simulate fetching user data via network service call with latency
                   (Thread/sleep 20)
-                  (-> observer (.onNext {:video-id videoId
-                                         ; 50/50 chance of giving back position 0 or 0-2500
-                                         :position (if (= 0 (rand-int 2)) 0 (rand-int 2500))}))
-                  (-> observer .onCompleted)
-                  (catch Exception e (-> observer (.onError e)))))]
+                  (-> s (.onNext {:video-id videoId
+                                  ; 50/50 chance of giving back position 0 or 0-2500
+                                  :position (if (= 0 (rand-int 2)) 0 (rand-int 2500))}))
+                  (-> s .onCompleted)
+                  (catch Exception e
+                    (-> s (.onError e)))))]
         ; a subscription that cancels the future if unsubscribed
-        (Subscriptions/create (rx/action [] (future-cancel f)))))))
+        (.add s (Subscriptions/create (rx/action [] (future-cancel f))))))))
 
-(defn getVideoMetadata [videoId, preferredLanguage]
+(defn getVideoMetadata
   "Asynchronously fetch movie metadata for a given language
   return Observable<Map>"
+  [videoId, preferredLanguage]
   (Observable/create
-    (rx/fn [observer]
+    (rx/action [^rx.Subscriber s]
       (let [f (future
+                (println "getVideoMetadata " videoId)
                 (try
                   ; simulate fetching video data via network service call with latency
                   (Thread/sleep 50)
                   ; contrived metadata for en-us or es-us
                   (if (= "en-us" preferredLanguage)
-                    (-> observer (.onNext {:video-id videoId
+                    (-> s (.onNext {:video-id videoId
                                            :title "House of Cards: Episode 1"
                                            :director "David Fincher"
                                            :duration 3365})))
                   (if (= "es-us" preferredLanguage)
-                    (-> observer (.onNext {:video-id videoId
+                    (-> s (.onNext {:video-id videoId
                                            :title "Cámara de Tarjetas: Episodio 1"
                                            :director "David Fincher"
                                            :duration 3365})))
-                  (-> observer .onCompleted)
-                  (catch Exception e (-> observer (.onError e))))) ]
+                  (-> s .onCompleted)
+                  (catch Exception e
+                    (-> s (.onError e))))) ]
         ; a subscription that cancels the future if unsubscribed
-        (Subscriptions/create (rx/action [] (future-cancel f)))))))
+        (.add s (Subscriptions/create (rx/action [] (future-cancel f))))))))
 
 
 (defn getVideoForUser [userId videoId]
@@ -188,8 +179,9 @@
   - user data
   return Observable<Map>"
   (let [user-observable           (-> (getUser userId)
-                                    (.map (rx/fn [user] {:user-name (:name user)
-                                                      :language (:preferred-language user)})))
+                                      (.map (rx/fn [user]
+                                              {:user-name (:name user)
+                                               :language (:preferred-language user)})))
         bookmark-observable       (-> (getVideoBookmark userId videoId)
                                     (.map (rx/fn [bookmark] {:viewed-position (:position bookmark)})))
         ; getVideoMetadata requires :language from user-observable so nest inside map function
@@ -218,8 +210,6 @@
 ;
 (comment
   (-> (getVideoForUser 12345 78965)
-    (.subscribe
-      (rx/action [x] (println "--- Object ---\n" x))
-      (rx/action [e] (println "--- Error ---\n" e))
-      (rx/action [] (println "--- Completed ---")))))
+      (.toBlockingObservable)
+      .single))