Managing Complexity with Concurrent Programming - Clojure Data Analysis - page 88

Monitoring processing with watchers
Another tool that Clojure provides for working with agents is watchers. These are just
functions that get a chance to peek at the agent's data. This happens after the validators
have successfully run and the new data is set as the agent's state. Because of the way it's
handled, the state may have changed again since then, but watchers give you the chance
to look at the data and track it separately.
This can help us keep an eye on the data as it's being processed. We can use it to log
progress, sample the data for manual validation, or a number of other tasks.
Getting ready
We'll need these dependencies:
(require '[clojure.java.io :as io]
'[clojure.data.csv :as csv]
'[clojure.string :as str])
(import '[java.lang Thread])
Also, we'll use the data iles from the Managing program complexity with agents recipe,
along with the binding to the list of those iles, data-files .
From Managing program complexity with STM , we'll use the lazy-read-csv and
->int functions.
How to do it…
In this recipe, we'll add a watcher to keep track of the number of rows that are converted,
and we'll add a lag that lets us know when processing is inished:
1. In order to convert the appropriate ields into rows, we'll need a couple of functions.
First, we'll have, row-ints , a data binding to indicate which ields need to be
converted to integers. Then, we'll deine try->int , which will normalize the input
and attempt to convert it to an integer. If it fails, it will return the original value.
Finally, coerce-row-ints will take a row, attempt to convert the rows with
integers, and send the results to the next agent:
(def row-ints [0 4])
(defn try->int [v]
(try
(->int (str/trim (str/replace v \| \space)))
(catch Exception ex
v)))
(defn coerce-row-ints [_ row indices sink]