 
 (defmethod mime-type-dispatch "application/pdf" [download]
   (counter "MIME types" "PDF")
-  (debug "mime-type-dispatch got PDF from " (:request_uri download)))
+  (debug "mime-type-dispatch got PDF from " (:request_uri download))
+  (handle-pdf download))
 
 (defmethod mime-type-dispatch "text/html" [download]
   (counter "MIME types" "HTML")
-  (debug "mime-type-dispatch got HTML from " (:request_uri download)))
+  (debug "mime-type-dispatch got HTML from " (:request_uri download))
+  (run-scrapers download))
 
 (defmethod mime-type-dispatch :default [download]
   (counter "MIME types" "unknown")

            [clojure.contrib.walk :as walk])
  (:use clojure.contrib.json.read
        org.altlaw.util.log)
  (:import (org.apache.commons.codec.binary Base64)))

(hadoop/setup-mapreduce)

(declare *reporter*)

(defn counter [group name]
  (.incrCounter *reporter* group name 1))

(def *docid-map* (ref {}))

(defn get-docid [uri]
  (get @*docid-map* uri))

(defn- decode-response-body [download]
  (assoc download :response_body_bytes
         (Base64/decodeBase64
          (.getBytes (:response_body_base64 download) "UTF-8"))))

;;; Step 2: dispatch on MIME type of the download



;;; MAPPER METHODS

(defn load-jruby []
  (alter-var-root #'*ruby*
                  (fn [_] (.getEngineByName (ScriptEngineManager.) "jruby")))
  (alter-var-root #'*ruby-context*
                  (fn [_] (.getContext *ruby*)))
  (.eval *ruby* (str "require 'org/altlaw/extract/scrape/run_scraper'"))
  (.eval *ruby* (str "$runner = RunScraper.new")))

(defn mapper-configure [this job]
  (let [cached (DistributedCache/getLocalCacheFiles job)
        path (first (filter #(= (.getName %) "docids.tsv.gz") cached))]
    (when (nil? path) (throw (RuntimeException. "No docid.tsv.gz in DistributedCache.")))
    (info "Loading docid map from " path)
    (alter-var-root #'*docid-map*
                    (fn [_] (tsv/load-tsv-map (str path))))
    (load-jruby)))

(defn mapper-map [this wkey wvalue output reporter]
    (let [input (walk/keywordize-keys (read-json-string (str wvalue)))]
      (debug "Called map with file position " wkey
             " and value " (logstr input))
      (binding [*reporter* reporter
                *raw-input* (str wvalue)]
        (doseq [[key value] (my-map input)]
          (.collect output (Text. (str key)) (Text. value))))))



