Understanding Transducers. See README below.

understanding-transducers.clj

(ns my-transducers.core
  (:require [clojure.core.async :as async]))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Understanding Transducers
;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; This is the source code for the blog post Understanding Transducers, found
;; here: http://elbenshira.com/blog/understanding-transducers
;;
;; Most of the code will run on any version of Clojure.
;; 
;; core.async section depends on Clojure 1.7.0-alpha1 or greater,
;; and core.async 0.1.338.0-5c5012-alpha or greater.


;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Power of reduce
;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(map inc (range 10))
; (1 2 3 4 5 6 7 8 9 10)

(filter even? '(1 2 3 4 5 6 7 8 9 10))
; (2 4 6 8 10)

(filter even? (map inc (range 10)))
; (2 4 6 8 10)

;; Notice that map and filter can be defined in terms of reduce

(defn map-inc-reducer
  [result input]
  (conj result (inc input)))

(reduce map-inc-reducer [] (range 10))
; [1 2 3 4 5 6 7 8 9 10]

(defn filter-even-reducer
  [result input]
  (if (even? input)
    (conj result input)
    result))

(reduce filter-even-reducer [] '(1 2 3 4 5 6 7 8 9 10))
; [2 4 6 8 10]

;; Extracting to a higher-order function

(defn map-reducer
  [f]
  (fn [result input]
    (conj result (f input))))

(reduce (map-reducer inc) [] (range 10))
; [1 2 3 4 5 6 7 8 9 10]

(defn filter-reducer
  [predicate]
  (fn [result input]
    (if (predicate input)
      (conj result input)
      result)))

(reduce (filter-reducer even?) [] '(1 2 3 4 5 6 7 8 9 10))
; [2 4 6 8 10]

(reduce
  (filter-reducer even?)
  []
  (reduce
    (map-reducer inc)
    []
    (range 10)))
; [2 4 6 8 10]

;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Another step in abstraction
;;;;;;;;;;;;;;;;;;;;;;;;;;;;

;; conj and + are reducing functions.
;; 
;; Reducing functions have the type
;; result, input -> result

(conj [1 2 3] 4)
; [1 2 3 4]

(+ 10 1)
; 11

;; Another higher-order, allowing user to pass reducing function

(defn mapping
  [f]
  (fn [reducing]
    (fn [result input]
      (reducing result (f input)))))

(defn filtering
  [predicate]
  (fn [reducing]
    (fn [result input]
      (if (predicate input)
        (reducing result input)
        result))))

(reduce
  ((filtering even?) conj)
  []
  (reduce
    ((mapping inc) conj)
    []
    (range 10)))
; [2 4 6 8 10]

;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Arriving at transducers
;;;;;;;;;;;;;;;;;;;;;;;;;;;;

;; Using reducing function.

(((mapping inc) conj) [] 1)
; [2]

(((mapping inc) conj) [2] 2)
; [2 3]

(((mapping inc) conj) [2 3] 3)
; [2 3 4]

(((filtering even?) conj) [2 4] 5)
; [2 4]

(((filtering even?) conj) [2 4] 6)
; [2 4 6]

;; This has the type: result, input -> result
((mapping inc) ((filtering even?) conj))

;; Cleaning up via comp.
(def xform
  (comp
    (mapping inc)
    (filtering even?)))

(reduce (xform conj) [] (range 10))
; [2 4 6 8 10]

(defn square [x] (* x x))

(def xform
  (comp
    (filtering even?) 
    (filtering #(< % 10))
    (mapping square)
    (mapping inc)))

(reduce (xform conj) [] (range 10))
; [1 5 17 37 65]

;; Turns out (mapping inc), (filtering even?) and xform are transducers.
;; They have the type: (result, input -> result) -> (result, input -> result).

;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; A more intuitive understanding
;;;;;;;;;;;;;;;;;;;;;;;;;;;;

((xform conj) [1 5 17] 12)
; [1 5 17]

((xform conj) [1 5 17] 6)
; [1 5 17 37]

(reduce (xform conj) [] (range 10))
; [1 5 17 37 65]

;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Transducers in core.async
;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn square [x] (* x x))

(def xform
  (comp
    (filter even?)
    (filter #(< % 10))
    (map square)
    (map inc)))

(def my-chan (async/chan 1 xform))

; Waiting for an item to print...
(async/take! my-chan println)

(async/put! my-chan 3)
; nothing printed to screen, since 3 is not even

(async/put! my-chan 4)
; "17" printed to screen, since 4 is even and less than 10


;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Problem sets
;;;;;;;;;;;;;;;;;;;;;;;;;;;;

;; ==========================
;; Problem: Write a transduce helper function
;; ==========================

(defn transduce
  [transducer reducing init coll]
  (reduce (transducer reducing) init coll))

(transduce (mapping inc) conj [] [1 2 3])

;; ==========================
;; Problem: The Caesar Cipher
;; ==========================
;;
;; - Filter out vowels and spaces,
;; - filter out upper-case characters,
;; - rotate all remaining characters via a [Caesar cipher](http://en.wikipedia.org/wiki/Caesar_cipher),
;; - and reduce the rotated characters into a map counting the number of occurrences of each character.

; (def vowels #{\a \e \i \o \u})
(def valid-chars (into #{} conj "bcdfghjklmnpqrstvwxyz"))

;; ASCII lower case characters range from 97 to 122
(defn rotate [cipher]
  (fn [c]
    (let [base-c (- (int c) 97)
          rotated-c (mod (+ base-c cipher) 26)]
      (char (+ rotated-c 97)))))

(defn caesar-reducing
  [result input]
  ; (fnil + 0) replaces use of #(+ (or % 0) 1), for when the key does not exist
  ; yet (and thus value is nil)
  (update-in result (str input) (fnil inc 0)))

(defn caesar-xform
  [cipher]
  (comp
    (filtering valid-chars)
    (filtering (comp not #(Character/isUpperCase %)))
    (mapping (rotate cipher))))

(defn caesar-count
  [string cipher]
  (transduce (caesar-xform cipher) caesar-reducing {} string))

(caesar-count "abc" 0)
; ⇒ {\c 1, \b 1}

(caesar-count "abc" 1)
; ⇒ {\d 1, \c 1}

(caesar-count "hello world" 0)
; ⇒ {\d 1, \r 1, \w 1, \l 3, \h 1}

(caesar-count "hello world" 13)
; ⇒ {\q 1, \e 1, \j 1, \y 3, \u 1}

;; ==========================
;; Problem: Write a mapcat transducer
;; ==========================

;; Assumes f returns a collection.
(defn mapcatting [f]
  (fn [reducing]
    (fn [result input]
      ;; mapping transducer does (reducing result (f input)). But for
      ;; mapcatting, we want to concatenate the collection returned from (f
      ;; input) into the result. The "concatenation" is defined by `reducing`.
      (reduce reducing result (f input)))))

(defn twins [x] [x x])

(mapcatting twins)

(((mapcatting twins) conj) [] 1)

(reduce ((mapcatting twins) conj) [] (range 10))

;; ==========================
;; Problem: Write a take transducer
;; ==========================

;; Only take n items
(defn taking [n]
  (let [c (atom 0)]
    (fn [reducing]
      (fn [result input]
        (if (< @c n)
          (do
            (swap! c inc)
            (reducing result input))
          result)))))

(((taking 3) conj) [] 1)

(reduce ((taking 3) conj) [] (range 10))

(def xform2
  (comp
    (filtering even?) 
    (mapping inc)  
    (mapcatting twins)
    (taking 10)))

(reduce (xform2 conj) [] (range 100))

understanding-transducers.md

Understanding Transducers

Understanding transducers by building them from scratch.

Goes along with the blog post Understanding Transducers.

Usage

Run lein repl. Then type the expressions found here.

Most of the code will run on any version of Clojure.

core.async section has these dependencies:

[org.clojure/clojure "1.7.0-alpha1"]
[org.clojure/core.async "0.1.338.0-5c5012-alpha"]

License

Copyright © 2014 Elben Shira

Distributed under the Eclipse Public License either version 1.0 or (at your option) any later version.

For taking transducer. I think the state should not be maintained in the taking fn, but in the fn returned by the taking function, like so

;; Only take n items
(defn taking [n]
  (fn [reducing]
    (let [c (atom 0)]
     (fn [result input]
       (if (< @c n)
         (do
           (swap! c inc)
           (reducing result input))
         result)))))

(def take3 (taking 3))
((take3 conj) [] 1)
(reduce (take3 conj) [] (range 10))

I think it would make sense to have the bottom-most form be (reduced result) rather than just returning result. Otherwise, even though you are only accumulating 3 results, the reduction continues for the entire length of the input collection by passing [0 1 2] as the result for the remaining 7 times.