    [clojure.tools.logging :as log])
   (:import
    [java.sql Connection DriverManager]))
+
+
+;; See:   "http://jdbc.postgresql.org/documentation/91/listennotify.html"
 
 (defonce registry         (atom {}))
 (defonce indexed-registry (atom {}))

             continue @(:continue conf)]
        (if continue
          (do
            ;; purely for side-effect
            (log/infof "db-watcher[%s] polling" dbname)
            (exec-sql "SELECT 1" [])
            (dispatch-notifications dbname (.getNotifications conn))
            (try
             (Thread/sleep sleep-time)
             (catch Exception e

   :password    (c :password)
   :port        (c :port)})

(defn process-row-changes [conf conn]
  ;; obtain mutex, nowait
  (let [dbname        (-> conf :dbconf :subname)]
   (jdbc/transaction
    (jdbc/do-commands "LOCK TABLE irmgard.process_log IN EXCLUSIVE MODE NOWAIT")
    (log/infof "obtained EXCLUSIVE lock on irmgard.process_log")
    ;; select and process the first block of records
    ;; TODO: make the LIMIT configurable
    (loop [recs (exec-sql "SELECT * FROM irmgard.row_changes ORDER BY event_id LIMIT 100" [])]
      (when (not (empty? recs))
        (doseq [rec recs]
          (doseq [[observer-name observer-fn] (find-listeners dbname (:schema_name rec) (:table_name rec))]
            (observer-fn rec))
          (jdbc/do-prepared "DELETE FROM irmgard.row_changes WHERE event_id=?" [(:event_id rec)]))
        (recur (exec-sql "SELECT * FROM irmgard.row_changes ORDER BY event_id LIMIT 100" []))))
    :ok)))

(defn dispatch-notifications [conf conn]
  ;; this select is purely for side-effect
  (exec-sql "SELECT 1" [])
  (let [dbname        (-> conf :dbname :subname)
        notifications (.getNotifications conn)]
    (when (not (nil? notifications))
      (log/infof "dispatch-notifications[%s]: %s notifications" dbname (count notifications))
      (process-row-changes conf conn))))

(defn config->jdbc-url [c]
  (format "jdbc:postgresql://%s:%s/%s"
