refactoring, writing is still not totally correct

Author: Christopher Merrick

README.md

@@ -1,14 +1,19 @@
-# deltalog
+# repwrite
 
-A Clojure library designed to ... well, that part is up to you.
+tap into your database's replication system and write all change data down to a file
 
-## Usage
+## usage
 
-FIXME
+_currently only support MySQL row-based replication_
 
-## License
+```bash
+lein run -m repwrite.core -f mysql-bin.000001 -n 8779 -p 5001
+```
 
-Copyright © 2014 FIXME
+in your my.cnf:
 
-Distributed under the Eclipse Public License either version 1.0 or (at
-your option) any later version.
+```
+server-id = 1234
+log-bin=mysql-bin
+binlog_format=row
+```

project.clj

@@ -10,9 +10,12 @@
                  [cascading/cascading-hadoop "2.2.0"]
                  [cascading/cascading-local "2.2.0"]
                  [cascading.gmarabout/cascading-json "0.0.3"]
+                 [cheshire "5.3.1"]
+                 [org.clojure/tools.cli "0.3.1"]
                  [cascading.avro/avro-scheme "2.1.1"
                   :exclusions [[org.slf4j/slf4j-log4j12]
                                [org.apache.hadoop/hadoop-core]
                                cascading/cascading-core]]
                  [org.apache.avro/avro "1.7.4" :exclusions [org.slf4j/slf4j-api]]
-                 [com.damballa/abracad "0.4.9"]])
+                 [com.damballa/abracad "0.4.9"]
+                 [midje "1.6.3"]])

src/deltalog/core.clj

@@ -1,4 +1,4 @@
-(ns deltalog.coerce
+(ns repwrite.coerce
   (:import [com.google.code.or.common.glossary.column
             Int24Column DecimalColumn DoubleColumn
             EnumColumn FloatColumn LongColumn]))

src/deltalog/coerce.clj renamed to src/repwrite/coerce.clj

@@ -0,0 +1,104 @@
+(ns repwrite.core
+  (:require [repwrite.coerce :refer [coerce]]
+            [repwrite.schema :as schema]
+            [cheshire.core :refer :all]
+            [abracad.avro :as avro]
+            [clojure.string :as string]
+            [clojure.tools.cli :refer [parse-opts]])
+  (:import [com.google.code.or OpenReplicator]
+           [com.google.code.or.binlog BinlogEventListener]
+           [com.google.code.or.binlog.impl.event WriteRowsEvent UpdateRowsEvent
+            TableMapEvent QueryEvent DeleteRowsEvent AbstractRowEvent])
+  (:gen-class))
+
+#_(defn root-dir [table-id] (str "example/raw/" table-id))
+
+(def root-dir (str "example/raw/"))
+
+(defn mk-dir [dir]
+  (.mkdir (java.io.File. dir))
+  dir)
+
+(defn write-avro [{:keys [table-id]} data]
+  (let [filename (str (mk-dir (root-dir table-id)) "/raw.avro")
+        schema (schema/tableid->schema table-id)]
+    (with-open [adf (if (.exists (clojure.java.io/as-file filename))
+                      (avro/data-file-writer filename)
+                      (avro/data-file-writer schema filename))]
+      (.append adf data))))
+
+(defn write-json [{:keys [data] :as event}]
+  (when (not (empty? data))
+    (let [filename (str root-dir "raw.json")]
+      (spit filename (str (generate-string data) "\n") :append true))))
+
+(defmulti parse-event-data class)
+(defmethod parse-event-data WriteRowsEvent
+  [e]
+  (map #(map coerce (.getColumns %)) (.getRows e)))
+(defmethod parse-event-data UpdateRowsEvent
+  [e]
+  (map #(map coerce (.getColumns (.getAfter %))) (.getRows e)))
+(defmethod parse-event-data DeleteRowsEvent
+  [e]
+  (map #(map coerce (.getColumns %)) (.getRows e)))
+(defmethod parse-event-data :default
+  [e]
+  (println (str e "\n"))
+  {})
+
+(defmulti parse-meta-data class)
+(defmethod parse-meta-data AbstractRowEvent
+  [e]
+  {:table-id (.getTableId e)})
+(defmethod parse-meta-data :default
+  [e]
+  {})
+
+(deftype MyListener []
+  BinlogEventListener
+  (onEvents
+    [this e]
+    (write-json (conj (parse-meta-data e)
+                     {:data (parse-event-data e)}))))
+
+(defn replicator
+  "get a replicator"
+  [{:keys [host port filename position] :as opts} listener]
+  (doto (OpenReplicator.)
+    (.setUser "replication-user")
+    (.setPassword "password")
+    (.setHost host)
+    (.setPort port)
+    (.setServerId 1234)
+    (.setBinlogFileName filename)
+    (.setBinlogPosition position)
+    (.setBinlogEventListener listener)))
+
+(def cli-options
+  [["-h" "--host HOST" "Replication master hostname or IP"
+    :default "127.0.0.1"]
+   ["-p" "--port PORT" "Replication master port number"
+    :default 3306
+    :parse-fn #(Integer/parseInt %)
+    :validate [#(< 0 % 0x10000) "Must be a number between 0 and 65536"]]
+   ["-f" "--filename FILENAME" "Binlog filename"]
+   ["-n" "--position POSITION" "Binlog position"
+    :parse-fn #(Integer/parseInt %)]])
+
+(defn error-msg [errors]
+  (str "The following errors occurred while parsing your command:\n\n"
+       (string/join \newline errors)))
+
+(defn exit [status msg]
+  (println msg)
+  (System/exit status))
+
+(defn -main [& args]
+  (let [{:keys [options arguments errors summary]} (parse-opts args cli-options)]
+    (cond
+     errors (exit 1 (error-msg errors))
+     (nil? (:filename options)) (exit 1 "A replication filename must be specified")
+     (nil? (:position options)) (exit 1 "A replication position must be specified"))
+    (mk-dir root-dir)
+    (.start (replicator options (MyListener.)))))

src/repwrite/core.clj

@@ -1,5 +1,5 @@
-(ns deltalog.mr
-  (:require [deltalog.schema :as schema])
+(ns repwrite.mr
+  (:require [repwrite.schema :as schema])
   (:import [org.apache.hadoop.fs Path FileSystem]
            [org.apache.hadoop.conf Configuration]
            [cascading.tap SinkMode]

src/deltalog/mr.clj renamed to src/repwrite/mr.clj

@@ -1,4 +1,4 @@
-(ns deltalog.schema
+(ns repwrite.schema
   (:require [abracad.avro :as avro])
   (:import [org.apache.avro Schema$Parser]))
 

src/deltalog/schema.clj renamed to src/repwrite/schema.clj

@@ -0,0 +1,62 @@
+(ns deltalog.core-test
+  (:use midje.sweet)
+  (:require [repwrite.core :refer :all])
+  (:import [com.google.code.or.binlog.impl.event WriteRowsEvent UpdateRowsEvent
+            TableMapEvent QueryEvent DeleteRowsEvent BinlogEventV4HeaderImpl]
+           [com.google.code.or.common.glossary Row Pair]
+           [com.google.code.or.common.glossary.column Int24Column]))
+
+(def header (doto (BinlogEventV4HeaderImpl.)
+              (.setTimestamp (long 10000))
+              (.setTimestampOfReceipt (long 10000))
+              (.setServerId (long 1))
+))
+
+(defn int-col [i] (Int24Column/valueOf i))
+
+(defn to-row [cols] (Row. cols))
+
+(defn to-pair [before after] (Pair. before after))
+
+(facts "row-based replication events"
+       (fact "it handles empty WriteRowsEvents"
+             (parse-event-data  (WriteRowsEvent. header)) => [])
+
+       (fact "it handles single-row WriteRowsEvents"
+             (let [e (doto (WriteRowsEvent. header)
+                       (.setRows [(to-row [(int-col 1)])]))]
+               (parse-event-data e) => [[1]]))
+
+       (fact "it handles multi-row WriteRowsEvents"
+             (let [e (doto (WriteRowsEvent. header)
+                       (.setRows [(to-row [(int-col 1)])
+                                  (to-row [(int-col 2)])]))]
+               (parse-event-data e) => [[1]
+                                   [2]]))
+
+       (fact "it handles single-row UpdateRowsEvents"
+             (let [e (doto (UpdateRowsEvent. header)
+                       (.setRows [(to-pair (to-row [(int-col 1)])
+                                           (to-row [(int-col 2)]))]))]
+               (parse-event-data e) => [[2]]))
+
+       (fact "it handles multi-row UpdateRowsEvents"
+             (let [e (doto (UpdateRowsEvent. header)
+                       (.setRows [(to-pair (to-row [(int-col 1)])
+                                           (to-row [(int-col 2)]))
+                                  (to-pair (to-row [(int-col 3)])
+                                           (to-row [(int-col 4)]))]))]
+               (parse-event-data e) => [[2]
+                                   [4]]))
+
+       (fact "it handles single-row DeleteRowsEvents"
+             (let [e (doto (DeleteRowsEvent. header)
+                       (.setRows [(to-row [(int-col 1)])]))]
+               (parse-event-data e) => [[1]]))
+
+       (fact "it handles multi-row DeleteRowsEvents"
+             (let [e (doto (DeleteRowsEvent. header)
+                       (.setRows [(to-row [(int-col 1)])
+                                  (to-row [(int-col 2)])]))]
+               (parse-event-data e) => [[1]
+                                   [2]])))