    :password    (c :password)
    :port        (c :port)})
 
-(defn dispatch-notifications [dbname notifications]
-  (when (not (nil? notifications))
-    (log/infof "dispatch-notifications[%s]: %s notifications" dbname (count notifications)))
-  )
+(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"

             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

    (DriverManager/getConnection url (:username c) (:password c))))

(comment
  (start-watcher :test1 (db-conn-info (config :db)))
  (stop-watcher :test1)


  (jdbc/with-connection  (db-conn-info (config :db))
    (exec-sql "SELECT now()" []))

  (register-listener :test-listener (:database (config :db)) "irmgard" "example_table"
                     (fn [r]
                       (log/infof "ROW CHANGE! %s" r)))

  (jdbc/with-connection  (db-conn-info (config :db))
    (process-row-changes
     {:dbconf (db-conn-info (config :db))}
     (jdbc/find-connection)))


  )