 (defn time [ms]
   (Time. ms))
 
-(defn streaming-context [conf batch-duration]
+(defn streaming-context 
+  "conf can be a SparkConf or JavaSparkContext"
+  [conf batch-duration]
   (JavaStreamingContext. conf (duration batch-duration)))
 
 (defn local-streaming-context [app-name duration]

      (.groupByKeyAndWindow (duration window-length) (duration slide-interval))
      (.map (function f/untuple))))

(defn reduce-by-window [dstream f window-length slide-interval]
  (.reduceByWindow dstream (function2 f) (duration window-length) (duration slide-interval)))

(defn reduce-by-key-and-window [dstream f window-length slide-interval]
  (-> dstream

(defmulti checkpoint (fn [context arg] (class arg)))
(defmethod checkpoint java.lang.String [streaming-context path] (.checkpoint streaming-context path))
(defmethod checkpoint java.lang.Long [dstream interval] (.checkpoint dstream (duration interval)))

(defn text-file-stream [context file-path]
  (.textFileStream context file-path))

(defn socket-text-stream [context ip port]
  (.socketTextStream context ip port))
