                           v)
                         d))]
     (->> data'
-         (graph->local command data')
-         (rx/finally
-           (println id "stop")))))
+      (graph->local command data')
+      (rx/finally
+        (println id "stop")))))
 
 (defn ^:private find-next-o [observable-lookup]
   (fn [command]

(defn observable->data
  ^Observable [^Observable o]
  (->> o
       (dereference-all)
       (rx/map (comp 'value pig/thaw-values))
       observable->clj))

(defn observable->raw-data
  ^Observable [^Observable o]
  (->> o
       (dereference-all)
       (rx/map pig/thaw-anything)
       observable->clj))

;; ********** Util **********


            [rx.lang.clojure.blocking :as rx-blocking]
            [pigpen.rx.extensions.core :refer [multicast]]
            [pigpen.extensions.io :refer [list-files]]
            [pigpen.extensions.core :refer [zipv]]
            [pigpen.pig :as pig])
  (:import [pigpen PigPenException]
           [org.apache.pig EvalFunc]
