threaded-style.clj

(def client-impl
  {:alter-topics* (fn [this topics]
                   (d/future
                     (->> (.alterConfigs this topics)
                          .all deref)))
   :create-topics* (fn [this topics]
                    (d/future
                      (->> (.createTopics this topics)
                           .all deref)))
   :delete-topics*  (fn [this topics]
                      (d/future
                        (->> (.deleteTopics this topics)
                             .all deref)))
   :describe-topics* (fn [this topics]
                       (d/future
                         (->> (.describeTopics this topics (DescribeTopicsOptions.))
                              .all deref)))
   :describe-configs* (fn [this configs]
                        (d/future
                          (->> (.describeConfigs this configs (DescribeConfigsOptions.))
                               .all .get)))
   :describe-cluster* (fn [this]
                       (d/future
                         (->> (.describeCluster this (DescribeClusterOptions.))
                              jd/datafy)))
   :list-topics* (fn [this]
                  (d/future
                    (->> (.listTopics this)
                         .names deref)))})

unthreaded-style.clj

(def client-impl
  {:alter-topics* (fn [this topics]
                    (d/future
                      @(.all (.alterConfigs this topics))))
   :create-topics* (fn [this topics]
                    (d/future
                      @(.all (.createTopics this topics))))
   :delete-topics*  (fn [this topics]
                      (d/future
                        @(.all (.deleteTopics this topics))))
   :describe-topics* (fn [this topics]
                       (d/future
                         @(.all (.describeTopics this topics (DescribeTopicsOptions.)))))
   :describe-configs* (fn [this configs]
                        (d/future
                          @(.all (.describeConfigs this configs (DescribeConfigsOptions.)))))
   :describe-cluster* (fn [this]
                        (d/future
                          (jd/datafy (.describeCluster this (DescribeClusterOptions.)))))
   :list-topics* (fn [this]
                   (d/future
                     @(.names (.listTopics this))))})

unthreaded-with-timeouts.clj

(def client-impl
  {:alter-topics* (fn [this topics]
                    (d/future
                      (deref (.all (.alterConfigs this topics)) 5000 :timeout))
   :create-topics* (fn [this topics]
                    (d/future
                      (deref (.all (.createTopics this topics)) 5000 :timeout)))
   :delete-topics*  (fn [this topics]
                      (d/future
                        (deref (.all (.deleteTopics this topics)) 5000 :timeout)))
   :describe-topics* (fn [this topics]
                       (d/future
                         (deref (.all (.describeTopics this topics (DescribeTopicsOptions.))) 5000 :timeout)))
   :describe-configs* (fn [this configs]
                        (d/future
                          (deref (.all (.describeConfigs this configs (DescribeConfigsOptions.))) 5000 :timeout)))

   :describe-cluster* (fn [this]
                        (d/future
                          ;; currently no way to specify a timeout for the `datafy` operation
                          ;; which for this result, dereferences a map of futures
                          (jd/datafy (.describeCluster this (DescribeClusterOptions.)))))
   :list-topics* (fn [this]
                   (d/future
                     (deref (.names (.listTopics this)) 5000 :timeout)))})