daveray

Say you have a function that filters a user's viewing history (sequence of maps, one for each "view" record) based on a bunch of criteria:

• User's country
• User's A/B test allocations
• Whether they watched a standalone movie or episode of a series.
• How much they watched
• Have they watched it before
• etc

These are all combined in a non-trivial way to decide whether a viewing record is included in the output.

daveray

(defn foo
  [context]  (wrap-router 
  (wrap-router (comp output-queue-map
                     (routing-table %)
                     :type)))

daveray

; https://github.com/Netflix/Hystrix

; Hystrix is a latency and fault tolerance library designed to 
; isolate points of access to remote systems, services and 3rd party 
; libraries, stop cascading failure and enable resilience in complex 
; distributed systems where failure is inevitable.

; Define a vanilla function
(defn search
  "Fault INtolerant search"

daveray

(ns com.netflix.indigena.rx.graph
  "This is an implementation namespace. Don't use it directly. Use the symbols
  in com.netflix.indigena.rx"
  (:require [clojure.set :as set]))

(def ^:private -ns- *ns*)
(set! *warn-on-reflection* true)

(defn ^:private ->let-o*-observable
  [^rx.Observable o n name]

daveray

Posted to Clojure Google Group, but it's not showing up there (https://groups.google.com/forum/#!topic/clojure/GITSKLmFM60), so...

In an effort to simplify the implementation and improve the RxJava experience for all JVM-based languages, we'll be dropping direct IFn support (and Groovy closure, Ruby proc, etc) in the next release. Instead, for each supported language (Scala, Groovy, and Clojure at the moment), we'll be providing a set of utility functions/macros/implicits to assist with interop. In the Clojure case, this will consist of a namespace, probably rx.lang.clojure, with the following helpers:

daveray

(defn chunk
  "Same as rx.Observable.merge(Observable<Observable<T>>) but the input Observables
  are \"chunked\" so that at most chunk-size of them are \"in flight\" at any given
  time.

  The order of the input Observables is not preserved.

  The main purpose here is to allow a large number of Hystrix observables to
  be processed in a controlled way so that the Hystrix execution queues aren't
  overwhelmed.

daveray

(defmacro ^:private reify-callable
  "Reify a bunch of Func* interfaces

    prefix  fully qualified interface name. numbers will be appended
    arities vector of desired arities
    f       the function to execute

  "
  [prefix arities f]
  (let [f-name (gensym "rc")]

daveray

rx/try
[(observable catch-clause* finally-clause?)]
Macro
  A macro that takes the place of onErrorResumeNext, doFinally, and friends. The resulting
  expression is an observable that acts as a pass through except for the way it handles
  errors.

    catch-clause  =>  (catch ClassName name expr*)
    finally-clause  =>  (finally expr*)

daveray

(defn lookup
  [dep-graph key]
  (let [{:keys [deps fn]} (dep-graph key)
        context           (map (partial lookup dep-graph) deps) ]
    (apply fn context)))

;###############################################################################
; Define some functions with deps encoded as metadata
(defn bar*
  [a b]

daveray

(defn lookup
  [dep-graph key]
  (let [var     (get dep-graph key)
        context (->> (for [d (-> var meta :deps)]
                       [d (lookup dep-graph d)])
                     (into {}))]
    (partial var context)))

;###############################################################################
; Define some functions with deps encoded as metadata