examples.fs.clj

(ns examples.fs
(:import
  (org.apache.log4j Level Logger)
  (java.io BufferedReader InputStream InputStreamReader))

(:use [clojure.contrib.test-is :only (deftest is)])
(:use [clojure.contrib.except :only (throwf throw-arg)])

(:require [clojure.contrib.duck-streams :as ds]
          [examples.imports :as import]))

(import/hadoop-conf)
(import/hadoop-dfs)
(import/hadoop-fs)
(import/hadoop-io)
(import/hadoop-mapreduce)

(defn local-dfs 
  []
  (let [conf (JobConf.)]
    (.set conf "fs.default.name" "file:///")
    (FileSystem/getLocal conf)))

(defn mini-dfs-cluster 
  []
  (.(Logger/getRootLogger) setLevel Level/ERROR)
  (System/setProperty "test.data" "/tmp") 
  (.(MiniDFSCluster. (Configuration.) 1 true nil) getFileSystem))

(def *fs* {:connection nil})
     
(defn find-fs
  []
  (:connection *fs*))

(defn fs
  []
  (or (find-fs)
      (throwf "not connected")))

(defn with-dfs
  [f body]
  (with-open [con (f)]
    (binding [*fs* (assoc *fs* :connection con)]
    (body))))

(defmacro with-local-dfs
  [& body]
  `(with-dfs local-dfs (fn [] ~@body)))
  
(defmacro with-mini-dfs-cluster
  [& body]
  `(with-dfs mini-dfs-cluster (fn [] ~@body)))

(defn as-path [x] (Path. x))

(defn exists 
  [src]
  (.exists (fs) (as-path src)))

(defn upload
  [src dest]
  (.copyFromLocalFile (fs) (as-path src) (as-path dest)))

(defn download
  [src dest]
  (.copyToLocal (fs) (as-path src) (as-path dest)))

(defn open 
  [src]
  (.open (fs) (as-path src)))

(defn reader 
  [src]
  (BufferedReader. (InputStreamReader. (open src))))

(defn read-line-seq
  [src]
  (ds/read-lines (reader src)))

(deftest exists-test 
  (with-local-dfs
    (is (exists "file.log"))))
	
(deftest read-line-seq-test
  (with-local-dfs
    (let [lines (read-line-seq "file.log")]
      (is (= "line1" (first lines)))
      (is (= "line2" (last lines))))))