provide fun-map integration (#18)

Fixes #17

README.md

@@ -30,6 +30,14 @@ Because all functions have their schemas incorporated, you can get the best deve
 
 Put `(malli.dev/start!)` in your `user.clj` will [enable clj-kondo](https://github.com/metosin/malli/blob/master/docs/function-schemas.md#tldr) to use the schemas when editing.
 
+## Single API
+
+Powered by [fun-map](https://github.com/robertluo/fun-map), you can use one single API for accessing a Kafka cluster without any further knowledge:
+
+ - [`kafka-cluster`](https://cljdoc.org/d/io.github.robertluo/waterfall/CURRENT/api/robertluo.waterfall#kafka-cluster)
+
+You can see an example in [this easy example notebook](notebook/easy.clj).
+
 ## API namespaces
 
   | namespace | Description |

deps.edn

@@ -11,7 +11,8 @@
                           com.github.seancorfield/expectations         {:mvn/version "2.0.160"}
                           com.taoensso/nippy                          {:mvn/version "3.3.0-alpha2"}
                           com.cognitect/transit-clj                   {:mvn/version "1.0.329"}
-                          metosin/malli                               {:mvn/version "0.9.2"}}}
+                          metosin/malli                               {:mvn/version "0.9.2"}
+                          io.github.robertluo/fun-map                 {:mvn/version "0.5.114"}}}
            :test
            {:extra-deps {lambdaisland/kaocha {:mvn/version "1.71.1119"}}
             :main-opts  ["-m" "kaocha.runner"]}

notebook/easy.clj

@@ -0,0 +1,47 @@
+;;# Easier API
+(ns easy
+  (:require [robertluo.waterfall :as wf]
+            [robertluo.waterfall.shape :as shape]))
+
+;; Suppose you just want to use Kafka right away, do not want to learn
+;; too much, and do not want to use manifold/kafka API, do not want know
+;; which namespaces to require, waterfall has a single function for you.
+
+;; ## One map to rule them all
+
+;; Suppose all you want is to publish message to a kafka topic:
+(def cluster
+  (wf/kafka-cluster
+   {::wf/nodes "localhost:9092"
+    ::wf/shapes [(shape/topic (constantly "sentence")) (shape/edn) (shape/value-only)]}))
+
+;; Declare a var (fn) allow us to put message.
+(def put! (::wf/put! cluster))
+
+;; Put clojure data onto it when needed!
+(put! {:words "Hello, world!"})
+
+;; When you have done using it, just close the cluster and it releases all resources.
+(.close cluster)
+
+;;If you want to consumer messages from kafka topics, a cluster need a little more configuration:
+(def consuming-cluster
+  (wf/kafka-cluster
+   {::wf/nodes "localhost:9092"
+    ::wf/shapes [(shape/topic (constantly "sentence")) (shape/edn) (shape/value-only)]
+    ::wf/consumer-config {:position :beginning}
+    ::wf/group-id "tester1"
+    ::wf/topics ["sentence"]}))
+
+;; Just refer to its `::wf/consume` key
+(def consume (::wf/consume consuming-cluster))
+
+;; Then register you interest on it, here, we just print every message out:
+(consume println)
+
+;;When done, close it. You might have to wait up to 10 seconds for it.
+(.close consuming-cluster)
+
+;;## where the power come from
+;; This easy integration are powered by [fun-map](https://github.com/robertluo/fun-map),
+;; so make suer you put it in your dependencies.

src/robertluo/waterfall.clj

@@ -1,9 +1,12 @@
 (ns robertluo.waterfall
   "API namespace for the library"
   (:require 
-   [robertluo.waterfall 
-    [core :as core]]
-   [manifold.stream :as ms]))
+   [robertluo.waterfall
+    [core :as core]
+    [util :as util]]
+   [manifold.stream :as ms]
+   [robertluo.waterfall.shape :as shape]
+   [robertluo.waterfall.util :as util]))
 
 (def schema
   "Schema for waterfall"
@@ -80,7 +83,82 @@
     (-> (ms/transform xform strm) (ms/connect sink))
     strm))
 
-(comment
+(defn kafka-cluster
+  "returns a convient life-cycle-map of a kafka cluster. It requires:
+    - `::nodes` kafka bootstrap servers
+    - `::shapes` shape of data, example: [(shape/topic (constantly \"sentence\"))(shape/edn)(shape/value-only)]
+   
+   If you just use it to publish message,
+    - optional `::producer-config` can specify additional kafka producer configuration.
+   
+   If you want to consume from topics:
+    - `::topics` the topic you want to subscribe to. example: [\"sentence\"]
+    - `::group-id` the group id for the message consumer
+    - optional `::source-xform` is a transducer to process message before consuming
+    - optional `::consumer-config` can specify additional kafka consumer configuration. With additions:
+      - `:position` either `:beginning` `:end`, none for commited position (default)
+      - `:poll-duration` for how long the consumer poll returns, is a Duration value, default 10 seconds
+   
+   The returned map has different level of key-values let you use:
+    - Highest level, no additional knowledge:
+      - For consumer: `::consume` a function, a one-arity (each message) function as its arg, returns nil.
+      - For producer: 
+        -`::put` a function with a message as its arg. e.g. ((::put return-map) {:a 3})
+        -`::put-all` a function with message sequence as its arg
+    - Mid level, if you need access of underlying manifold sink/source.
+      - `::source` a manifold source.
+      - `::sink` a manifold sink.
+    - Lowest level, if you want to access kafka directly:
+      - `::consumer` a Kafka consumer.
+      - `::producer` a Kafka message producer.
+   "
+  {:malli/schema
+   [:=> schema [:cat [:map {:closed true}
+                      [::nodes ::nodes]
+                      [::shapes [:vector [:fn shape/shape?]]]
+                      [::producer-config {:optional true} ::producer-config]
+                      [::group-id {:optional true} :string]
+                      [::topics {:optional true} [:vector :string]]
+                      [::source-xform {:optional true} fn?]]]
+    :map]}
+  [kafka-conf-map]
+  (util/optional-require
+   [robertluo.fun-map :as fm :refer [fw fnk]]
+   (merge
+    (fm/life-cycle-map
+     {::producer (fnk [::nodes ::producer-config]
+                      (let [prod (producer nodes (or producer-config {}))]
+                        (fm/closeable prod #(ms/close! prod))))
+      ::consumer (fnk [::nodes ::group-id ::topics ::consumer-config]
+                      (assert (and group-id topics) "Has to provide group-id and topics")
+                      (let [cmer (consumer nodes group-id topics (or consumer-config {}))]
+                        (fm/closeable cmer #(ms/close! cmer))))
+      ::sink (fnk [::producer ::shapes]
+                 (-> producer ignore (xform-sink (comp (map (shape/serializer shapes))))))
+      ::source (fnk [::consumer ::shapes ::source-xform]
+                    (->> (comp (map (shape/deserializer shapes)) (or source-xform (map identity)))
+                         (xform-source consumer)))
+      ::put! (fnk [::sink] (partial ms/put! sink))
+      ::put-all (fnk [::sink] (partial ms/put-all! sink))
+      ::consume (fnk [::source] #(ms/consume % source))})
+    kafka-conf-map)
+   (throw (ClassNotFoundException. "Need io.github.robertluo/fun-map library in the classpath"))))
+
+(comment 
   (require '[malli.dev]) 
   (malli.dev/start!)
+  (malli.dev/stop!)
+  (def clu
+    (kafka-cluster 
+     {::nodes "localhost:9092"
+      ::shapes [(shape/topic (constantly "sentence")) (shape/edn) (shape/value-only)]
+      ::consumer-config {:position :beginning}
+      ::group-id "tester1"
+      ::topics ["sentence"]
+      ::source-xform (map identity)}))
+  (def put! (::put! clu))
+  (put! "Hello, world")
+  (def consume (::consume clu))
+  (consume println) 
+  (.close clu) 
   )

src/robertluo/waterfall/shape.clj

@@ -99,6 +99,15 @@
   [shapes]
   (->> (map :des shapes) (reverse) (apply comp)))
 
+(defn shapes
+  "construct shapes in order."
+  ([top-shapes]
+   (shapes top-shapes :edn))
+  ([top-shapes bytes-shape]
+   (shapes top-shapes bytes-shape (value-only)))
+  ([top-shapes bytes-shape kv-shape]
+   (-> (concat top-shapes [bytes-shape kv-shape]) vector)))
+
 (comment
   (def shapes [(edn) (key-value identity)])
   ((serializer shapes) [:foo "bar"])