cr comments

src/main/clojure/aerospike_clj/aerospike_record.clj

@@ -15,15 +15,15 @@
 (defn record->map [^Record record]
   (and record
        (let [bins    ^Map (.bins record)
-             payload (and bins
-                          (if (single-bin? bins)
-                            ;; single bin record
-                            (utils/desanitize-bin-value (.get bins ""))
-                            ;; multiple-bin record
-                            (reduce-kv (fn [m k v]
-                                         (assoc m k (utils/desanitize-bin-value v)))
-                                       {}
-                                       bins)))]
+             payload (when (some? bins)
+                       (if (single-bin? bins)
+                         ;; single bin record
+                         (utils/desanitize-bin-value (.get bins ""))
+                         ;; multiple-bin record
+                         (reduce-kv (fn [m k v]
+                                      (assoc m k (utils/desanitize-bin-value v)))
+                                    {}
+                                    bins)))]
          (->AerospikeRecord
            payload
            ^Integer (.generation ^Record record)

src/main/clojure/aerospike_clj/client.clj

@@ -144,7 +144,8 @@
                      (let [transcoder (:transcoder conf identity)]
                        (-> res
                            record/record->map
-                           transcoder))) completion-executor)
+                           transcoder)))
+                   completion-executor)
           (register-events client-events :read index start-time conf))))
 
   (get-single-no-meta [this index set-name]
@@ -191,15 +192,15 @@
   (exists-batch [_this indices conf]
     (let [op-future  (p/deferred)
           start-time (System/nanoTime)
+          transcoder (:transcoder conf identity)
           indices    (utils/v->array Key (mapv #(pt/create-key (:index %) dbns (:set %)) indices))]
       (.exists ^AerospikeClient client
                ^EventLoop (.next ^EventLoops el)
                (AsyncExistsArrayListener. op-future)
                ^BatchPolicy (:policy conf)
                ^"[Lcom.aerospike.client.Key;" indices)
       (-> op-future
-          (p/then' vec completion-executor)
-          (p/then' (:transcoder conf identity))
+          (p/then' (comp transcoder vec) completion-executor)
           (register-events client-events :exists-batch nil start-time conf))))
 
   pt/AerospikeWriteOps
@@ -297,7 +298,10 @@
           set-name
           conf))
 
-  (touch [_this index set-name expiration]
+  (touch [this index set-name expiration]
+    (pt/touch this index set-name expiration {}))
+
+  (touch [_this index set-name expiration conf]
     (let [op-future  (p/deferred)
           start-time (System/nanoTime)]
       (.touch ^AerospikeClient client
@@ -307,7 +311,7 @@
               ^Key (pt/create-key index dbns set-name))
       (-> op-future
           (p/then' identity completion-executor)
-          (register-events client-events :touch index start-time {}))))
+          (register-events client-events :touch index start-time conf))))
 
   pt/AerospikeDeleteOps
   (delete [this index set-name]
@@ -374,15 +378,15 @@
                        (->> batch-records
                             (utils/v->array BatchRecord)
                             (Arrays/asList)))
-          start-time (System/nanoTime)]
+          start-time (System/nanoTime)
+          transcoder (:transcoder conf identity)]
       (.operate ^AerospikeClient client
                 ^EventLoop (.next ^EventLoops el)
                 ^BatchOperateListListener (AsyncBatchOperateListListener. op-future)
                 ^BatchPolicy policy
                 ^List batch-list)
       (-> op-future
-          (p/then' #(mapv batch-record->map %) completion-executor)
-          (p/then' (:transcoder conf identity))
+          (p/then' (comp transcoder #(mapv batch-record->map %)) completion-executor)
           (register-events client-events :batch-operate nil start-time conf))))
 
 

src/main/clojure/aerospike_clj/listeners.clj

@@ -1,10 +1,10 @@
 (ns aerospike-clj.listeners
   (:require [promesa.core :as p]
             [aerospike-clj.aerospike-record :as record])
-  (:import [java.util List Map]
-           [com.aerospike.client Key Record AerospikeException AerospikeException$QueryTerminated]
-           [com.aerospike.client.listener RecordListener WriteListener DeleteListener
-                                          ExistsListener BatchListListener RecordSequenceListener InfoListener ExistsArrayListener BatchOperateListListener]))
+  (:import (java.util List Map)
+           (com.aerospike.client Key Record AerospikeException AerospikeException$QueryTerminated)
+           (com.aerospike.client.listener RecordListener WriteListener DeleteListener
+                                          ExistsListener BatchListListener RecordSequenceListener InfoListener ExistsArrayListener BatchOperateListListener)))
 
 (deftype AsyncExistsListener [op-future]
   ExistsListener

src/main/clojure/aerospike_clj/protocols.clj

@@ -86,7 +86,9 @@
     "Add bins to an existing record without modifying old data. The `new-data` must be a
     Clojure map.")
 
-  (touch [this index set-name expiration]
+  (touch
+    [this index set-name expiration]
+    [this index set-name expiration conf]
     "Updates the TTL of the record stored under at `index` to `expiration` seconds from now.
     Expects records to exist."))
 

test/aerospike_clj/integration/aerospike_setup.clj

@@ -2,30 +2,27 @@
   (:require [clj-test-containers.core :as tc]))
 
 (def ^:private ^:const container-port 3000)
-(def ^:private container (atom nil))
+(def ^:dynamic *container* nil)
 
 (defn get-host-and-port []
-  (format "%s:%s" (:host @container) (get (:mapped-ports @container) container-port)))
+  (format "%s:%s" (:host *container*) (get (:mapped-ports *container*) container-port)))
 
 (defn- start-container []
   (->> {:image-name    "aerospike:ee-6.2.0.3"
-       :exposed-ports [container-port]
-       :wait-for      {:wait-strategy   :log
-                       :message         "objects: all 0 master 0 prole 0 non-replica 0"
-                       :times           1
-                       :startup-timeout 15}}
-      tc/create
-      tc/start!
-      (reset! container)))
+        :exposed-ports [container-port]
+        :wait-for      {:wait-strategy   :log
+                        :message         "objects: all 0 master 0 prole 0 non-replica 0"
+                        :times           1
+                        :startup-timeout 15}}
+       tc/create
+       tc/start!))
 
 (defn- stop! []
-  (tc/stop! @container)
-  (reset! container nil))
+  (tc/stop! *container*))
 
 (defn with-aerospike [test-fn]
-  (start-container)
-  (try
-    (test-fn)
-    (finally
-      (stop!))))
-
+  (binding [*container* (start-container)]
+    (try
+      (test-fn)
+      (finally
+        (stop!)))))