proceesor, storage: better error handling; no interrupting pipeline

- print clear stack trace using timbre when capturing exceptions
- don't pass nil to url/normalize; if url/normalize gets a nil, instead
of NPE, it now returns an empty string
- capture exceptions in processor pipeline to keep it from dying
- http.downloader now checks response status instead of blindly
downloading anything

4 files changed, 37 insertions, 19 deletions

diff --git a/src/weepinbell/http/downloader.clj b/src/weepinbell/http/downloader.clj
index 5acfcc2..d58501b 100644
--- a/src/weepinbell/http/downloader.clj
+++ b/src/weepinbell/http/downloader.clj
@@ -11,17 +11,25 @@
           parent-dir (.getParentFile file)]
       (when (and parent-dir (not (.exists parent-dir)))
         (.mkdirs parent-dir))
-      (let [response (client/get! url {:as :stream})]
-        (timbre/info "Downloading [" url "] -> [" dest-path "]")
-        ;; write to output stream
-        (with-open [in (:body response)
-                    out (io/output-stream file)]
-          (io/copy in out))
-        (timbre/info "Download success. File size:" (.length file) "bytes")
-        true))
+      (let [response (client/get! url {:as :stream})
+            status (:status response)]
+        (if (and status (> status 199) (< status 300))
+          (do 
+            (timbre/info "Downloading [" url "] -> [" dest-path "]")
+            ;; write to output stream
+            (with-open [in (:body response)
+                        out (io/output-stream file)]
+              (io/copy in out))
+            (timbre/info "Download success. File size:" (.length file) "bytes")
+            true)
+          (do 
+            (timbre/warn "Download fail. Status: " status)
+            (when-let [error (:error response)]
+              (timbre/error error))))))
     ;; when download failed
     (catch Exception e
       (timbre/error "Download error:" (.getMessage e))
+      (timbre/debug e)
       (io/delete-file (io/file dest-path))
       false)))
 
diff --git a/src/weepinbell/processor/facade.clj b/src/weepinbell/processor/facade.clj
index 9acea16..224c739 100644
--- a/src/weepinbell/processor/facade.clj
+++ b/src/weepinbell/processor/facade.clj
@@ -35,14 +35,18 @@
           (do

             (timbre/info "webmention found in queue, processing.")

             ((:on-receive callbacks) {:webmention {:id id :source s :target t}})

-            (-> {:id id

-                 :wm wm

-                 :ok? true}

-                (check-storage storage)

-                verify

-                (clean storage callbacks)

-                (wrap config)

-                (write storage callbacks))

+            (try

+              (-> {:id id

+                   :wm wm

+                   :ok? true}

+                  (check-storage storage)

+                  verify

+                  (clean storage callbacks)

+                  (wrap config)

+                  (write storage callbacks))

+              (catch Exception e

+                (timbre/error "unexpceted exception in pipeline: " (.getMessage e))

+                (timbre/debug e)))

             (recur (<! wmc))))))

     ;; return a function that stops the loop 

     (fn [] (put! wmc {:stop? true}))))

diff --git a/src/weepinbell/storage/edn_storage.clj b/src/weepinbell/storage/edn_storage.clj
index 6a06c28..42ddc71 100644
--- a/src/weepinbell/storage/edn_storage.clj
+++ b/src/weepinbell/storage/edn_storage.clj
@@ -16,8 +16,12 @@
 
 ;; predicates used for filtering webmention
 (defn- st-pair-equal-pred [source target wm]
-  (and (url= source (get-in wm [:webmention :source]))
-       (url= target (get-in wm [:webmention :target]))))
+  (let [wm-source (get-in wm [:webmention :source])
+        wm-target (get-in wm [:webmention :target])]
+    (and (some? source) (some? target)
+         (some? wm-source) (some? wm-target)
+         (url= source wm-source)
+         (url= target wm-target))))
 (defn- id-equal-pred [id wm]
   (= id (get-in wm [:webmention :id])))
 (defn- get-predicate [source target]
diff --git a/src/weepinbell/utils/url.clj b/src/weepinbell/utils/url.clj
index 03e7da8..5daf292 100644
--- a/src/weepinbell/utils/url.clj
+++ b/src/weepinbell/utils/url.clj
@@ -50,7 +50,9 @@
         query  (normalize-query (.getQuery base))]

       (str scheme "://" auth path

         (when (seq query) (str "?" query))))

-    ;; fall back to original string if syntax is invalid

+    ;; fallback to empty string is url is nil

+    (catch NullPointerException _ "")

+    ;; fallback to original string if syntax is invalid

     (catch URISyntaxException _ url)))

 

 (defn url= [& urls]