   (write [this writer value])
   (close-writer [this writer]))
 
-;; unsubscribe doesn't seem to work with this version of rx
-;; This caps the maximum number of records read from a file/stream
-(def ^:dynamic *max-load-records* nil)
-
 (defmethod graph->local :load
   [{:keys [location], :as command} _]
   (let [local-loader (load command)
         ^Observable o (->> (rx/observable*
-                             (fn [^Subscriber o]
+                             (fn [^Subscriber s]
                                (let [cancel (atom false)
                                      read-count (atom 0)]
-                                 (.add o
-                                       (reify Subscription
-                                         (unsubscribe [this]
-                                           (reset! cancel true))))
+                                 (.add s
+                                   (reify Subscription
+                                     (unsubscribe [this]
+                                       (reset! cancel true))))
                                  (future
                                    (try
                                      (println "Start reading from " location)
                                      (doseq [file (locations local-loader)
-                                             :while (not @cancel)
-                                             :while (< @read-count (or *max-load-records* Double/POSITIVE_INFINITY))]
+                                             :while (not @cancel)]
                                        (let [reader (init-reader local-loader file)]
                                          (doseq [value (read local-loader reader)
-                                                 :while (not @cancel)
-                                                 :while (< @read-count (or *max-load-records* Double/POSITIVE_INFINITY))]
+                                                 :while (not @cancel)]
                                            (swap! read-count inc)
-                                           (.onNext o value))
+                                           (rx/on-next s value))
                                          (close-reader local-loader reader)))
-                                     (.onCompleted o)
-                                     (catch Exception e (.onError o e))))
-                                 )))
-                           (rx/finally
-                             (println "Stop reading from " location)))]
+                                     (rx/on-completed s)
+                                     (catch Exception e (rx/on-error s e)))))))
+                        (rx/finally
+                          (println "Stop reading from " location)))]
     (.observeOn o (Schedulers/io))))
 
 (defmethod graph->local :store

        :desc (int (- r))))))

(defmethod graph->local :order [{:keys [sort-keys]} data]
  (let [^Observable data (first data)]
    (-> data
      (.toSortedList (rx-interop/fn* (partial pig-compare sort-keys)))
      (.flatMap (rx-interop/fn* rx/seq->o)))))

(defmethod graph->local :rank [{:keys [sort-keys]} data]
  (let [^Observable data (first data)]
    (if (not-empty sort-keys)
      (-> data
        (.toSortedList (rx-interop/fn* (partial pig-compare sort-keys)))
        (.flatMap (rx-interop/fn* #(rx/seq->o (map-indexed (fn [i v] (assoc v '$0 i)) %)))))
      (let [i (atom -1)]
        (->> data
             (rx/map (fn [v] (assoc v '$0 (swap! i inc)))))))))

;; ********** Filter **********

(defmethod graph->local :filter [{:keys [code]} data]
  (let [^Observable data (first data)]
    (rx/filter (fn [values]
                 (eval-code code values))
               data)))

(defmethod graph->local :filter-native [{:keys [fields expr]} data]
  (if-not expr (first data)
    (let [^Observable data (first data)
          f (eval `(fn [{:syms ~fields}] ~expr))]
      (rx/filter (fn [values] (f values))
                 data))))

(defmethod graph->local :distinct [_ data]
  (let [^Observable data (first data)
        seen (atom [false #{}])]
    (rx/filter (fn [values]
                 (let [[c _] (swap! seen (fn [[_ s]] [(contains? s values)
                                                      (conj s values)]))]
                   (not c)))
               data)))

(defmethod graph->local :limit [{:keys [n]} data]
  (let [^Observable data (first data)]
    (rx/take n data)))

(defmethod graph->local :sample [{:keys [p]} data]
  (let [^Observable data (first data)]
    (rx/filter (fn [values] (< (rand) p))
               data)))

;; ********** Join **********

(defmethod graph->local :union [_ ^List data]
  (apply rx/merge data))

(defn ^:private graph->local-group [{:keys [ancestors keys join-types fields]} data]
  (->
    ^List
    (mapv (fn [a k ^Observable d j]
            (.flatMap d (rx-interop/fn [values]
                          ;; This selects all of the fields that are in this relation
                          (let [^Iterable result
                                (for [[[r] v :as f] (next fields)
                                      :when (= r a)]
                                  {:values (pig/tuple (values v))
                                   ;; This is to emulate the way pig handles nils
                                   ;; This changes a nil values into a relation specific nil value
                                   :key (mapv #(or (values %) (keyword (name a) "nil")) k)
                                   :relation f
                                   :required (= j :required)})]
                            (rx/seq->o result)))))
         ancestors keys data join-types)
    (Observable/merge)
    (.groupBy (rx-interop/fn* :key))
    (.flatMap (rx-interop/fn [^Observable o]
                (-> ^Observable o
                  (.groupBy (rx-interop/fn* :relation))
                  (.flatMap (rx-interop/fn [%]
                              (-> ^Observable %
                                   (.toList)
                                   (.map (rx-interop/fn [v]
                                             [(if (keyword? (.getKey ^GroupedObservable %))
                                                nil
                                                (.getKey ^GroupedObservable %))
                                              (apply pig/bag (mapv :values v))])))))
                  ;; Start with the group key. If it's a single value, flatten it.
                  ;; Keywords are the fake nils we put in earlier
                  (.reduce {(first fields) (let [k (.getKey ^GroupedObservable o)
                                                 k (if (= 1 (count k)) (first k) k)
                                                 k (if (keyword? k) nil k)]
                                             k)}
                    (rx-interop/fn [values [k v]]
                      (assoc values k v))))))
    ;; TODO This is a bad way to do inner groupings
    (.filter (rx-interop/fn [g]
               (every? identity
                 (map (fn [a [k] j]
                        (or (= j :optional)
                            (contains? g [[a] k])))
                      ancestors
                      keys
                      join-types))))))

(defn ^:private graph->local-group-all [{:keys [fields]} data]
  (->
    ^List
    (mapv (fn [[r v] ^Observable d]
            (.map d (rx-interop/fn [values]
                      ;; TODO clean up pig dereferencing
                      (let [v' (v values)]
                        {:values (if (instance? Tuple v') v' (pig/tuple v'))
                         :relation [r v]}))))
         (next fields) data)
    (Observable/merge)
    (.groupBy (rx-interop/fn* :relation))
    (.flatMap (rx-interop/fn [^Observable %]
                (-> %
                    (.toList)
                    (.map (rx-interop/fn [v]
                            [(.getKey ^GroupedObservable %)
                             (apply pig/bag (mapv :values v))])))))
    (.reduce {(first fields) nil}
      (rx-interop/fn [values [k v]]
        (assoc values k v)))
    (.filter (rx-interop/fn [v]
               (< 1 (count v))))))

(defmethod graph->local :group [{:keys [keys] :as command} data]
  (if (= keys [:pigpen.raw/group-all])


(defn dereference-all ^Observable [^Observable o]
  (rx/map #(->> %
             (map (fn [[k v]]
                    [k (dereference v)]))
             (into {}))
          o))

(defn observable->clj [^Observable o]
