bump version to 0.9.0-SNAPSHOT

various optimisations/fixes

.gitignore

@@ -9,3 +9,5 @@ pom.xml.asc
 /.nrepl-port
 .hgignore
 .hg/
+.lsp/
+.clj-kondo/

CHANGELOG.md

@@ -1,6 +1,22 @@
 # Change Log
 All notable changes to this project will be documented in this file. This change log follows the conventions of [keepachangelog.com](http://keepachangelog.com/).
 
+## Unreleased
+
+## 0.8.0 - 2024-02-07
+### Added
+- Support for passing `:realise-fn` to avoid lazy response (thanks @jimpil)
+
+### Changed
+- Remove reflective calls
+
+## 0.7.0 - 2022-10-04
+### Added
+- Support reading dates as instanceres (thanks @henryw374)
+- Support data literal for mongo id (thanks @henryw374)
+- Support for implicit transactions (thanks @AdamClements)
+- Support for aggregation pipeline in find-one-and-update (requires 4.2+, thanks @jacobemcken)
+
 ## 0.6.0 - 2020-01-10
 ### Added
 - Support for bulk-write

README.md

@@ -5,7 +5,7 @@
 [![cljdoc badge](https://cljdoc.org/badge/mongo-driver-3/mongo-driver-3)](https://cljdoc.org/d/mongo-driver-3/mongo-driver-3/CURRENT)
 
 
-A Mongo client for clojure, lightly wrapping 3.11+ versions of the [MongoDB Java Driver](https://mongodb.github.io/mongo-java-driver/)
+A Mongo client for clojure, lightly wrapping 3.11/4.0+ versions of the [MongoDB Java Driver](https://mongodb.github.io/mongo-java-driver/)
 
 In general, it will feel familiar to users of mongo clients like [monger](https://github.com/michaelklishin/monger).
 Like our HTTP/2 client [hato](https://github.com/gnarroway/hato), the API is designed to be idiomatic and to make common 
@@ -23,7 +23,7 @@ It was developed with the following goals:
 
 ## Status
 
-mongo-driver-3 is used in production, but is also under development and the API may change slightly.
+mongo-driver-3 is used in production, and the existing public API will be maintained.
 Please try it out and raise any issues you may find.
 
 ## Usage
@@ -32,10 +32,10 @@ For Leinengen, add this to your project.clj:
 
 ```clojure
 ;; The underlying driver -- any newer version can also be used
-[org.mongodb/mongodb-driver-sync "3.11.2"]
+[org.mongodb/mongodb-driver-sync "4.11.1"]
 
 ;; This wrapper library
-[mongo-driver-3 "0.6.0"]
+[mongo-driver-3 "0.8.0"]
 ```
 
 ## Getting started
@@ -104,6 +104,10 @@ As an example:
 ;; Find a single document or return nil
 (mc/find-one db "test" {:v "world"} {:keywordize? false})
 ; => {"v" "world"}
+
+;; Avoid laziness in queries 
+(mc/find db "test" {} {:realise-fn (partial into [])}
+; => [...]
 ```
 
 While most options are supported directly, sometimes you may need to some extra control.
@@ -142,18 +146,18 @@ except each operation is defined as a 2-tuple rather than a map.
 
 ```clojure
 ;; Execute a mix of operations in one go
-(,c/bulk-write [[:insert-one {:document {:a 1}}]
-               [:delete-one {:filter {:a 1}}]
-               [:delete-many {:filter {:a 1}}]
-               [:update-one {:filter {:a 1} :update {:$set {:a 2}}}]
-               [:update-many {:filter {:a 1} :update {:$set {:a 2}}}]
-               [:replace-one {:filter {:a 1} :replacement {:a 2}}]])
+(bulk-write [[:insert-one {:document {:a 1}}]
+             [:delete-one {:filter {:a 1}}]
+             [:delete-many {:filter {:a 1}}]
+             [:update-one {:filter {:a 1} :update {:$set {:a 2}}}]
+             [:update-many {:filter {:a 1} :update {:$set {:a 2}}}]
+             [:replace-one {:filter {:a 1} :replacement {:a 2}}]])
 ; => a BulkWriteResult
              
 ;; Each operation can take the same options as their respective functions
-(mc/bulk-write [[:update-one {:filter {:a 1} :update {:$set {:a 2}} :upsert? true}]
-               [:update-many {:filter {:a 1} :update {:$set {:a 2}} :upsert? true}]
-               [:replace-one {:filter {:a 1} :replacement {:a 2} :upsert? true}]])
+(bulk-write [[:update-one {:filter {:a 1} :update {:$set {:a 2}} :upsert? true}]
+             [:update-many {:filter {:a 1} :update {:$set {:a 2}} :upsert? true}]
+             [:replace-one {:filter {:a 1} :replacement {:a 2} :upsert? true}]])
 ```
 
 ### Using transactions
@@ -171,8 +175,23 @@ use `with-open` so the session is closed after both successful and failed transa
     (fn []
       (mc/insert-one my-db "coll" {:name "hello"} {:session s})
       (mc/insert-one my-db "coll" {:name "world"} {:session s}))))
+
+;; There is also a helper method to make this easier,
+;; where it is not necessary to manually open or pass a session:
+(mg/with-implicit-transaction
+  {:client client}
+  (fn []
+    (mc/insert-one my-db "coll" {:name "hello"}) 
+    (mc/insert-one my-db "coll" {:name "world"})))
 ```
 
+## Development
+
+1. Run mongo (e.g. via docker):
+    - `docker run -it --rm -p 27017:27017 mongo`
+2. Run tests
+    - `lein test`
+
 ## License
 
 Released under the MIT License: http://www.opensource.org/licenses/mit-license.php

project.clj

@@ -1,5 +1,5 @@
-(defproject mongo-driver-3 "0.6.0"
-  :description "A Clojure wrapper for the Java MongoDB driver 3.11+."
+(defproject mongo-driver-3 "0.9.0-SNAPSHOT"
+  :description "A Clojure wrapper for the Java MongoDB driver 3.11/4.0+."
   :url "https://github.com/gnarroway/mongo-driver-3"
   :license {:name         "The MIT License"
             :url          "http://opensource.org/licenses/mit-license.php"
@@ -10,5 +10,5 @@
                                     :sign-releases false}]]
   :plugins [[lein-cljfmt "0.6.4"]]
 
-  :profiles {:dev {:dependencies [[org.clojure/clojure "1.10.1"]
-                                  [org.mongodb/mongodb-driver-sync "3.11.2"]]}})
+  :profiles {:dev {:dependencies [[org.clojure/clojure "1.11.1"]
+                                  [org.mongodb/mongodb-driver-sync "4.11.1"]]}})

src/data_readers.clj

@@ -0,0 +1 @@
+{mongo/id mongo-driver-3.data-literals/mongo-id}

src/mongo_driver_3/client.clj

@@ -1,6 +1,7 @@
 (ns mongo-driver-3.client
   (:refer-clojure :exclude [find])
-  (:require [mongo-driver-3.model :as m])
+  (:require [mongo-driver-3.model :as m]
+            [mongo-driver-3.iterable :as iterable])
   (:import (com.mongodb.client MongoClients MongoClient ClientSession MongoDatabase TransactionBody)
            (com.mongodb ConnectionString ClientSessionOptions TransactionOptions)
            (java.util.concurrent TimeUnit)))
@@ -15,16 +16,15 @@
   `connection-string` is a mongo connection string, e.g. mongodb://localhost:27107
 
   If a connecting string is not passed in, it will connect to the default localhost instance."
-  ([] (MongoClients/create))
-  ([^String connection-string]
-   (MongoClients/create connection-string)))
+  (^MongoClient []                          (MongoClients/create))
+  (^MongoClient [^String connection-string] (MongoClients/create connection-string)))
 
 (defn get-db
   "Gets a database by name
 
   `client` is a MongoClient, e.g. resulting from calling `connect`
   `name` is the name of the database to get."
-  [^MongoClient client ^String name]
+  ^MongoDatabase [^MongoClient client ^String name]
   (.getDatabase client name))
 
 (defn close
@@ -41,16 +41,20 @@
   - `opts` (optional), a map of:
     - `:name-only?` returns just the string names
     - `:keywordize?` keywordize the keys of return results, default: true. Only applicable if `:name-only?` is false.
-    - `:raw?` return the mongo iterable directly instead of processing into a seq, default: false
+    - `:raw?` return the mongo iterable directly instead of processing into a clj data-structure, default: false
+    - `:realise-fn` how to realise the MongoIterable, default: `clojure.core/sequence` (i.e. lazily)
     - `:session` a ClientSession"
   ([^MongoDatabase db] (list-collections db {}))
-  ([^MongoDatabase db {:keys [raw? keywordize? ^ClientSession session] :or {keywordize? true}}]
+  ([^MongoDatabase db {:keys [raw? keywordize? ^ClientSession session realise-fn] 
+                       :or {keywordize? true
+                            realise-fn sequence}}]
    (let [it (if session
               (.listCollections db session)
               (.listCollections db))]
-     (if-not raw?
-       (map #(m/from-document % keywordize?) (seq it))
-       it))))
+     (if raw?
+       it
+       (realise-fn ;; accomodate users who don't want to use lazy-seqs
+         (iterable/documents it keywordize?))))))
 
 (defn list-collection-names
   "Lists collection names in a database, returning as a seq of strings unless otherwise configured.
@@ -66,9 +70,9 @@
    (let [it (if-let [^ClientSession session (:session opts)]
               (.listCollectionNames db session)
               (.listCollectionNames db))]
-     (if-not (:raw? opts)
-       (seq it)
-       it))))
+     (if (:raw? opts)
+       it
+       (seq it)))))
 
 (defn ->TransactionOptions
   "Coerces options map into a TransactionOptions. See `start-session` for usage."
@@ -130,11 +134,14 @@
   Ensure `session` is passed as an option to each operation.
 
   e.g.
+
+  ```clojure
   (with-open [s (start-session client)]
     (with-transaction s
       (fn []
         (insert-one my-db \"coll\" {:name \"hello\"} {:session s})
-        (insert-one my-db \"coll\" {:name \"world\"} {:session s}))))"
+        (insert-one my-db \"coll\" {:name \"world\"} {:session s}))))
+  ```"
   ([^ClientSession session body] (with-transaction session body {}))
   ([^ClientSession session body opts]
    (.withTransaction session
@@ -142,6 +149,37 @@
                        (execute [_] (body)))
                      (->TransactionOptions opts))))
 
+
+(def ^:dynamic *session* nil)
+
+(defn with-implicit-transaction
+  "Automatically sets the session / transaction for all mongo operations
+   within the scope, using a dynamic binding.
+
+   The first argument is an options map with keys:
+
+   - `:client` a MongoClient (mandatory)
+   - `:transaction-opts` (see `->TransactionOptions` for keys)
+   - `:session-opts` (see `start-session` for details)
+
+   The second argument `body` is a fn with one or more mongo operations in it.
+   e.g.
+
+   ```
+   (mg/with-implicit-transaction
+     {:client client}
+     (fn []
+       (mc/insert-one my-db \"coll\" {:name \"hello\"})
+       (mc/insert-one my-db \"coll\" {:name \"world\"})))
+   ```"
+  [{:keys [^MongoClient client transaction-opts session-opts] :or {transaction-opts {} session-opts {}}} body]
+  (with-open [^ClientSession session (start-session client session-opts)]
+    (binding [*session* session]
+      (with-transaction
+        *session*
+        body
+        transaction-opts))))
+
 ;;; Utility
 
 (defn connect-to-db

src/mongo_driver_3/collection.clj

@@ -1,6 +1,8 @@
 (ns mongo-driver-3.collection
   (:refer-clojure :exclude [find empty? drop])
-  (:require [mongo-driver-3.model :refer :all])
+  (:require [mongo-driver-3.model :refer :all]
+            [mongo-driver-3.client :refer [*session*]]
+            [mongo-driver-3.iterable :as iterable])
   (:import (com.mongodb MongoNamespace)
            (com.mongodb.client MongoDatabase MongoCollection ClientSession)
            (com.mongodb.client.model IndexModel)
@@ -61,22 +63,27 @@
     - `:batch-size` Documents to return per batch, e.g. 1
     - `:bypass-document-validation?` Boolean
     - `:keywordize?` keywordize the keys of return results, default: true
+    - `:realise-fn` how to realise the MongoIterable, default: `clojure.core/sequence` (i.e. lazily)
     - `:raw?` return the mongo AggregateIterable directly instead of processing into a seq, default: false
     - `:session` a ClientSession"
   ([^MongoDatabase db coll pipeline]
    (aggregate db coll pipeline {}))
   ([^MongoDatabase db coll pipeline opts]
-   (let [{:keys [^ClientSession session allow-disk-use? ^Integer batch-size bypass-document-validation? keywordize? raw?] :or {keywordize? true raw? false}} opts
-         it (cond-> (if session
-                      (.aggregate (collection db coll opts) session ^List (map document pipeline))
-                      (.aggregate (collection db coll opts) ^List (map document pipeline)))
-              (some? allow-disk-use?) (.allowDiskUse allow-disk-use?)
-              (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?)
-              batch-size (.batchSize batch-size))]
+   (let [{:keys [^ClientSession session allow-disk-use? ^Integer batch-size bypass-document-validation? keywordize? raw? realise-fn]
+          :or {keywordize? true
+               realise-fn sequence}} opts
+         ^ClientSession session (or session *session*)
+         it                     (cond-> (if session
+                                          (.aggregate (collection db coll opts) session ^List (map document pipeline))
+                                          (.aggregate (collection db coll opts) ^List (map document pipeline)))
+                                  (some? allow-disk-use?) (.allowDiskUse allow-disk-use?)
+                                  (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?)
+                                  batch-size (.batchSize batch-size))]
 
-     (if-not raw?
-       (map (fn [x] (from-document x keywordize?)) (seq it))
-       it))))
+     (if raw?
+       it
+       (realise-fn ;; accomodate users who don't want to use lazy-seqs
+        (iterable/documents it keywordize?))))))
 
 (defn bulk-write
   "Executes a mix of inserts, updates, replaces, and deletes.
@@ -102,7 +109,7 @@
    (bulk-write db coll operations {}))
   ([^MongoDatabase db coll operations opts]
    (let [opts' (->BulkWriteOptions opts)]
-     (if-let [session (:session opts)]
+     (if-let [session (or (:session opts) *session*)]
        (.bulkWrite (collection db coll opts') ^ClientSession session ^List (map write-model operations))
        (.bulkWrite (collection db coll opts') (map write-model operations))))))
 
@@ -130,7 +137,7 @@
    (count-documents db coll q {}))
   ([^MongoDatabase db coll q opts]
    (let [opts' (->CountOptions opts)]
-     (if-let [session (:session opts)]
+     (if-let [session (or (:session opts) *session*)]
        (.countDocuments (collection db coll opts) session (document q) opts')
        (.countDocuments (collection db coll opts) (document q) opts')))))
 
@@ -150,7 +157,7 @@
   ([^MongoDatabase db coll q]
    (delete-one db coll q {}))
   ([^MongoDatabase db coll q opts]
-   (if-let [session (:session opts)]
+   (if-let [session (or (:session opts) *session*)]
      (.deleteOne (collection db coll opts) session (document q) (->DeleteOptions opts))
      (.deleteOne (collection db coll opts) (document q) (->DeleteOptions opts)))))
 
@@ -170,7 +177,7 @@
   ([^MongoDatabase db coll q]
    (delete-many db coll q {}))
   ([^MongoDatabase db coll q opts]
-   (if-let [session (:session opts)]
+   (if-let [session (or (:session opts) *session*)]
      (.deleteMany (collection db coll opts) session (document q) (->DeleteOptions opts))
      (.deleteMany (collection db coll opts) (document q) (->DeleteOptions opts)))))
 
@@ -188,25 +195,30 @@
     - `:sort` document representing sort order, e.g. {:timestamp -1}
     - `:projection` document representing fields to return, e.g. {:_id 0}
     - `:keywordize?` keywordize the keys of return results, default: true
+    - `:realise-fn` how to realise the MongoIterable, default: `clojure.core/sequence` (i.e. lazily)
     - `:raw?` return the mongo FindIterable directly instead of processing into a seq, default: false
     - `:session` a ClientSession
 
   Additionally takes options specified in `collection`."
   ([^MongoDatabase db coll q]
    (find db coll q {}))
-  ([^MongoDatabase db coll q opts]
-   (let [{:keys [limit skip sort projection ^ClientSession session keywordize? raw?] :or {keywordize? true raw? false}} opts]
-     (let [it (cond-> (if session
-                        (.find (collection db coll opts) session (document q))
-                        (.find (collection db coll opts) (document q)))
-                limit (.limit limit)
-                skip (.skip skip)
-                sort (.sort (document sort))
-                projection (.projection (document projection)))]
+  ([^MongoDatabase db coll q {:keys [limit skip sort projection ^ClientSession session keywordize? raw? realise-fn]
+                              :or {keywordize? true 
+                                   realise-fn sequence}
+                              :as opts}]
+   (let [^ClientSession session (or session *session*)
+         it                     (cond-> (if session
+                                          (.find (collection db coll opts) session (document q))
+                                          (.find (collection db coll opts) (document q)))
+                                  limit (.limit limit)
+                                  skip (.skip skip)
+                                  sort (.sort (document sort))
+                                  projection (.projection (document projection)))]
 
-       (if-not raw?
-         (map (fn [x] (from-document x keywordize?)) (seq it))
-         it)))))
+     (if raw? 
+       it
+       (realise-fn ;; accomodate users who don't want to use lazy-seqs
+         (iterable/documents it keywordize?))))))
 
 (defn find-one
   "Finds a single document and returns it as a clojure map, or nil if not found.
@@ -225,7 +237,10 @@
   - `db` is a MongoDatabase
   - `coll` is a collection name
   - `q` is a map representing a query.
-  - `update` is a map representing an update. The update to apply must include only update operators.
+  - `update` is either a map representing a document update or a vector
+    representing an 'aggregation pipeline'. A document update must include only
+    update operators, while an 'aggregation pipeline' can contain multiple
+    stages of `$set`, `$unset` and `$replaceWith`.
   - `opts` (optional), a map of:
     - `:upsert?` whether to insert a new document if nothing is found, default: false
     - `:return-new?` whether to return the document after update (insead of its state before the update), default: false
@@ -241,10 +256,16 @@
    (find-one-and-update db coll q update {}))
   ([^MongoDatabase db coll q update opts]
    (let [{:keys [keywordize? ^ClientSession session] :or {keywordize? true}} opts
-         opts' (->FindOneAndUpdateOptions opts)]
-     (-> (if session
-           (.findOneAndUpdate (collection db coll opts) session (document q) (document update) opts')
-           (.findOneAndUpdate (collection db coll opts) (document q) (document update) opts'))
+         ^ClientSession session (or session *session*)
+         opts'                  (->FindOneAndUpdateOptions opts)]
+     (-> (if (instance? List update)
+           (let [pipeline ^List (map document update)]
+             (if session
+               (.findOneAndUpdate (collection db coll opts) session (document q) pipeline opts')
+               (.findOneAndUpdate (collection db coll opts) (document q) pipeline opts')))
+           (if session
+             (.findOneAndUpdate (collection db coll opts) session (document q) (document update) opts')
+             (.findOneAndUpdate (collection db coll opts) (document q) (document update) opts')))
          (from-document keywordize?)))))
 
 (defn find-one-and-replace
@@ -271,7 +292,8 @@
    (find-one-and-replace db coll q doc {}))
   ([^MongoDatabase db coll q doc opts]
    (let [{:keys [keywordize? session] :or {keywordize? true}} opts
-         opts' (->FindOneAndReplaceOptions opts)]
+         session (or session *session*)
+         opts'   (->FindOneAndReplaceOptions opts)]
      (-> (if session
            (.findOneAndReplace (collection db coll opts) session (document q) (document doc) opts')
            (.findOneAndReplace (collection db coll opts) (document q) (document doc) opts'))
@@ -297,7 +319,7 @@
    (insert-one db coll doc {}))
   ([^MongoDatabase db coll doc opts]
    (let [opts' (->InsertOneOptions opts)]
-     (if-let [session (:session opts)]
+     (if-let [session (or (:session opts) *session*)]
        (.insertOne (collection db coll opts) session (document doc) opts')
        (.insertOne (collection db coll opts) (document doc) opts')))))
 
@@ -322,7 +344,7 @@
    (insert-many db coll docs {}))
   ([^MongoDatabase db coll docs opts]
    (let [opts' (->InsertManyOptions opts)]
-     (if-let [^ClientSession session (:session opts)]
+     (if-let [^ClientSession session (or (:session opts) *session*)]
        (.insertMany (collection db coll opts) session ^List (map document docs) opts')
        (.insertMany (collection db coll opts) ^List (map document docs) opts')))))
 
@@ -346,7 +368,7 @@
   ([^MongoDatabase db coll q doc]
    (find-one-and-replace db coll q doc {}))
   ([^MongoDatabase db coll q doc opts]
-   (if-let [^ClientSession session (:session opts)]
+   (if-let [^ClientSession session (or (:session opts) *session*)]
      (.replaceOne (collection db coll opts) session (document q) (document doc) (->ReplaceOptions opts))
      (.replaceOne (collection db coll opts) (document q) (document doc) (->ReplaceOptions opts)))))
 
@@ -370,7 +392,7 @@
   ([^MongoDatabase db coll q update]
    (update-one db coll q update {}))
   ([^MongoDatabase db coll q update opts]
-   (if-let [^ClientSession session (:session opts)]
+   (if-let [^ClientSession session (or (:session opts) *session*)]
      (.updateOne (collection db coll opts) session (document q) (document update) (->UpdateOptions opts))
      (.updateOne (collection db coll opts) (document q) (document update) (->UpdateOptions opts)))))
 
@@ -394,7 +416,7 @@
   ([^MongoDatabase db coll q update]
    (update-many db coll q update {}))
   ([^MongoDatabase db coll q update opts]
-   (if-let [^ClientSession session (:session opts)]
+   (if-let [^ClientSession session (or (:session opts) *session*)]
      (.updateMany (collection db coll opts) session (document q) (document update) (->UpdateOptions opts))
      (.updateMany (collection db coll opts) (document q) (document update) (->UpdateOptions opts)))))
 
@@ -480,7 +502,7 @@
    (create-indexes db coll indexes {}))
   ([^MongoDatabase db coll indexes opts]
    (->> indexes
-        (map (fn [x] (IndexModel. (document (:keys x)) (->IndexOptions x))))
+        (mapv (fn [x] (IndexModel. (document (:keys x)) (->IndexOptions x))))
         (.createIndexes (collection db coll opts)))))
 
 (defn list-indexes
@@ -488,5 +510,7 @@
   ([^MongoDatabase db coll]
    (list-indexes db coll {}))
   ([^MongoDatabase db coll opts]
-   (->> (.listIndexes (collection db coll opts))
-        (map #(from-document % true)))))
+   (let [it (.listIndexes (collection db coll opts))
+         realise-fn (:realise-fn opts sequence)]
+     (realise-fn 
+       (iterable/documents it true)))))

src/mongo_driver_3/data_literals.clj

@@ -0,0 +1,36 @@
+(ns mongo-driver-3.data-literals
+  (:import (org.bson.types ObjectId)
+           (java.io Writer)
+           (java.util Date)
+           (java.nio ByteBuffer)))
+
+
+(defmethod print-method ObjectId [^ObjectId c ^Writer w] (.write w (str "#mongo/id " \" (.toHexString c) \")))
+(defmethod print-dup ObjectId    [^ObjectId c ^Writer w] (.write w (str "#mongo/id " \" (.toHexString c) \")))
+
+(defprotocol AsObjectId 
+  (oid-from [this]))
+
+(extend-protocol AsObjectId 
+  (Class/forName "[B") 
+  (oid-from [this] (ObjectId. ^bytes this))
+  nil 
+  (oid-from [_]    (ObjectId.))
+  String 
+  (oid-from [this] (ObjectId. this))
+  Date 
+  (oid-from [this] (ObjectId. this))
+  ByteBuffer
+  (oid-from [this] (ObjectId. this))
+  )
+
+
+
+(defn mongo-id ;; https://mongodb.github.io/mongo-java-driver/4.8/apidocs/bson/org/bson/types/ObjectId.html
+  (^ObjectId []  (ObjectId.))
+  (^ObjectId [o] (oid-from o))
+  (^ObjectId [o1 o2] 
+   (if (and (int? o1) 
+            (int? o2))
+     (ObjectId. (int o1) (int o2))
+     (ObjectId. ^Date o1 (int o2)))))

src/mongo_driver_3/iterable.clj

@@ -0,0 +1,8 @@
+(ns mongo-driver-3.iterable 
+  (:require [mongo-driver-3.model :as m]))
+
+(defn documents 
+  "Given a MongoIterable <it>, returns an eduction which will
+   eventually yield all the documents (per `m/from-document`)."
+  [it keywordize?]
+  (eduction (map #(m/from-document % keywordize?)) it))

src/mongo_driver_3/model.clj

@@ -3,8 +3,8 @@
            (org.bson Document)
            (java.util.concurrent TimeUnit)
            (com.mongodb WriteConcern ReadPreference ReadConcern)
-           (clojure.lang Ratio Keyword Named IPersistentMap)
-           (java.util Collection List)
+           (clojure.lang Ratio Named IPersistentMap)
+           (java.util Collection List Date)
            (org.bson.types Decimal128)))
 
 (set! *warn-on-reflection* true)
@@ -14,6 +14,12 @@
 (defprotocol ConvertToDocument
   (^Document document [input] "Convert from clojure to Mongo Document"))
 
+(defn read-dates-as-instants! []
+  (extend-protocol ConvertToDocument
+    Date
+    (document [input _]
+      (.toInstant ^Date input))))
+
 (extend-protocol ConvertToDocument
   nil
   (document [_]
@@ -23,24 +29,22 @@
   (document [^Ratio input]
     (double input))
 
-  Keyword
-  (document [^Keyword input]
-    (.getName input))
-
   Named
   (document [^Named input]
     (.getName input))
 
   IPersistentMap
   (document [^IPersistentMap input]
-    (let [o (Document.)]
-      (doseq [[k v] input]
-        (.append o (document k) (document v)))
-      o))
+    (reduce-kv
+     (fn [^Document doc k v]
+       (doto doc 
+         (.append (document k) (document v))))
+     (Document.)
+     input))
 
   Collection
   (document [^Collection input]
-    (map document input))
+    (mapv document input))
 
   Object
   (document [input]
@@ -63,16 +67,18 @@
 
   List
   (from-document [^List input keywordize?]
-    (vec (map #(from-document % keywordize?) input)))
+    (mapv #(from-document % keywordize?) input))
 
   Document
   (from-document [^Document input keywordize?]
-    (reduce (if keywordize?
-              (fn [m ^String k]
-                (assoc m (keyword k) (from-document (.get input k) true)))
-              (fn [m ^String k]
-                (assoc m k (from-document (.get input k) false))))
-            {} (.keySet input))))
+    (persistent!
+     (reduce (if keywordize?
+               (fn [m ^String k]
+                 (assoc! m (keyword k) (from-document (.get input k) true)))
+               (fn [m ^String k]
+                 (assoc! m k (from-document (.get input k) false))))
+             (transient {})
+             (.keySet input)))))
 
 ;;; Config
 
@@ -86,24 +92,25 @@
 
 (defn ->ReadConcern
   "Coerce `rc` into a ReadConcern if not nil. See `collection` for usage."
-  [{:keys [read-concern]}]
+  ^ReadConcern [{:keys [read-concern]}]
   (when read-concern
     (if (instance? ReadConcern read-concern)
       read-concern
-      (or (kw->ReadConcern read-concern) (throw (IllegalArgumentException.
-                                                 (str "No match for read concern of " (name read-concern))))))))
+      (or (kw->ReadConcern read-concern)
+          (throw (IllegalArgumentException.
+                  (str "No match for read concern of " (name read-concern))))))))
 
 (defn ->ReadPreference
   "Coerce `rp` into a ReadPreference if not nil. See `collection` for usage."
-  [{:keys [read-preference]}]
+  ^ReadPreference [{:keys [read-preference]}]
   (when read-preference
     (if (instance? ReadPreference read-preference)
       read-preference
       (ReadPreference/valueOf (name read-preference)))))
 
-(defn ^WriteConcern ->WriteConcern
+(defn ->WriteConcern
   "Coerces write-concern related options to a WriteConcern. See `collection` for usage."
-  [{:keys [write-concern ^Integer write-concern/w ^Long write-concern/w-timeout-ms ^Boolean write-concern/journal?]}]
+  ^WriteConcern [{:keys [write-concern ^Integer write-concern/w ^Long write-concern/w-timeout-ms ^Boolean write-concern/journal?]}]
   (when (some some? [write-concern w w-timeout-ms journal?])
     (let [^WriteConcern wc (when write-concern
                              (if (instance? WriteConcern write-concern)
@@ -114,17 +121,17 @@
         w-timeout-ms (.withWTimeout w-timeout-ms (TimeUnit/MILLISECONDS))
         (some? journal?) (.withJournal journal?)))))
 
-(defn ^BulkWriteOptions ->BulkWriteOptions
+(defn ->BulkWriteOptions
   "Coerce options map into BulkWriteOptions. See `bulk-write` for usage."
-  [{:keys [bulk-write-options bypass-document-validation? ordered?]}]
+  ^BulkWriteOptions  [{:keys [bulk-write-options bypass-document-validation? ordered?]}]
   (let [^BulkWriteOptions opts (or bulk-write-options (BulkWriteOptions.))]
     (cond-> opts
       (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?)
       (some? ordered?) (.ordered ordered?))))
 
-(defn ^CountOptions ->CountOptions
+(defn ->CountOptions
   "Coerce options map into CountOptions. See `count-documents` for usage."
-  [{:keys [count-options hint limit max-time-ms skip]}]
+  ^CountOptions [{:keys [count-options hint limit max-time-ms skip]}]
   (let [^CountOptions opts (or count-options (CountOptions.))]
     (cond-> opts
       hint (.hint (document hint))
@@ -132,15 +139,15 @@
       max-time-ms (.maxTime max-time-ms (TimeUnit/MILLISECONDS))
       skip (.skip skip))))
 
-(defn ^DeleteOptions ->DeleteOptions
+(defn ->DeleteOptions
   "Coerce options map into DeleteOptions. See `delete-one` and `delete-many` for usage."
-  [{:keys [delete-options]}]
+  ^DeleteOptions [{:keys [delete-options]}]
   (let [^DeleteOptions opts (or delete-options (DeleteOptions.))]
     opts))
 
-(defn ^FindOneAndReplaceOptions ->FindOneAndReplaceOptions
+(defn ->FindOneAndReplaceOptions
   "Coerce options map into FindOneAndReplaceOptions. See `find-one-and-replace` for usage."
-  [{:keys [find-one-and-replace-options upsert? return-new? sort projection]}]
+  ^FindOneAndReplaceOptions [{:keys [find-one-and-replace-options upsert? return-new? sort projection]}]
   (let [^FindOneAndReplaceOptions opts (or find-one-and-replace-options (FindOneAndReplaceOptions.))]
     (cond-> opts
       (some? upsert?) (.upsert upsert?)
@@ -148,9 +155,9 @@
       sort (.sort (document sort))
       projection (.projection (document projection)))))
 
-(defn ^FindOneAndUpdateOptions ->FindOneAndUpdateOptions
+(defn ->FindOneAndUpdateOptions
   "Coerce options map into FindOneAndUpdateOptions. See `find-one-and-update` for usage."
-  [{:keys [find-one-and-update-options upsert? return-new? sort projection]}]
+  ^FindOneAndUpdateOptions  [{:keys [find-one-and-update-options upsert? return-new? sort projection]}]
   (let [^FindOneAndUpdateOptions opts (or find-one-and-update-options (FindOneAndUpdateOptions.))]
     (cond-> opts
       (some? upsert?) (.upsert upsert?)
@@ -169,40 +176,40 @@
       (some? sparse?) (.sparse sparse?)
       (some? unique?) (.unique unique?))))
 
-(defn ^InsertManyOptions ->InsertManyOptions
+(defn ->InsertManyOptions
   "Coerce options map into InsertManyOptions. See `insert-many` for usage."
-  [{:keys [insert-many-options bypass-document-validation? ordered?]}]
+  ^InsertManyOptions [{:keys [insert-many-options bypass-document-validation? ordered?]}]
   (let [^InsertManyOptions opts (or insert-many-options (InsertManyOptions.))]
     (cond-> opts
       (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?)
       (some? ordered?) (.ordered ordered?))))
 
-(defn ^InsertOneOptions ->InsertOneOptions
+(defn ->InsertOneOptions
   "Coerce options map into InsertOneOptions. See `insert-one` for usage."
-  [{:keys [insert-one-options bypass-document-validation?]}]
+  ^InsertOneOptions [{:keys [insert-one-options bypass-document-validation?]}]
   (let [^InsertOneOptions opts (or insert-one-options (InsertOneOptions.))]
     (cond-> opts
       (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?))))
 
-(defn ^ReplaceOptions ->ReplaceOptions
+(defn ->ReplaceOptions
   "Coerce options map into ReplaceOptions. See `replace-one` and `replace-many` for usage."
-  [{:keys [replace-options upsert? bypass-document-validation?]}]
+  ^ReplaceOptions [{:keys [replace-options upsert? bypass-document-validation?]}]
   (let [^ReplaceOptions opts (or replace-options (ReplaceOptions.))]
     (cond-> opts
       (some? upsert?) (.upsert upsert?)
       (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?))))
 
-(defn ^UpdateOptions ->UpdateOptions
+(defn ->UpdateOptions
   "Coerce options map into UpdateOptions. See `update-one` and `update-many` for usage."
-  [{:keys [update-options upsert? bypass-document-validation?]}]
+  ^UpdateOptions [{:keys [update-options upsert? bypass-document-validation?]}]
   (let [^UpdateOptions opts (or update-options (UpdateOptions.))]
     (cond-> opts
       (some? upsert?) (.upsert upsert?)
       (some? bypass-document-validation?) (.bypassDocumentValidation bypass-document-validation?))))
 
-(defn ^CreateCollectionOptions ->CreateCollectionOptions
+(defn ->CreateCollectionOptions
   "Coerce options map into CreateCollectionOptions. See `create` usage."
-  [{:keys [create-collection-options capped? max-documents max-size-bytes]}]
+  ^CreateCollectionOptions [{:keys [create-collection-options capped? max-documents max-size-bytes]}]
   (let [^CreateCollectionOptions opts (or create-collection-options (CreateCollectionOptions.))]
     (cond-> opts
       (some? capped?) (.capped capped?)

test/mongo_driver_3/client_test.clj

@@ -1,8 +1,9 @@
 (ns mongo-driver-3.client-test
   (:require [clojure.test :refer :all]
             [mongo-driver-3.client :as mg]
-            [mongo-driver-3.collection :as mc])
-  (:import (com.mongodb.client MongoClient MongoDatabase MongoIterable ListCollectionsIterable)
+            [mongo-driver-3.collection :as mc]
+            [mongo-driver-3.model :as m])
+  (:import (com.mongodb.client MongoClient MongoDatabase MongoIterable ListCollectionsIterable ClientSession)
            (java.util UUID)
            (com.mongodb ClientSessionOptions ReadConcern ReadPreference)
            (java.util.concurrent TimeUnit)))
@@ -83,23 +84,44 @@
 
   (deftest ^:integration test-with-transaction
     (let [db (new-db @client)
-          _ (mc/create db "test")]
+          _  (mc/create db "test")]
       (with-open [session (mg/start-session @client)]
         (is (= 2 (mg/with-transaction session
+                                      (fn []
+                                        (mc/insert-one db "test" {:a 1} {:session session})
+                                        (mc/insert-one db "test" {:a 2} {:session session})
+                                        (mc/count-documents db "test" {} {:session session}))))))))
+
+  (deftest ^:integration test-with-implicit-transaction
+    (testing "passing"
+      (let [db (new-db @client)
+            _  (mc/create db "test")]
+        (is (= 2 (mg/with-implicit-transaction
+                   {:client @client}
                    (fn []
-                     (mc/insert-one db "test" {:a 1} {:session session})
-                     (mc/insert-one db "test" {:a 2} {:session session})
-                     (mc/count-documents db "test" {} {:session session}))))))))
+                     (mc/insert-one db "test" {:a 1})
+                     (mc/insert-one db "test" {:a 2})
+                     (mc/count-documents db "test" {})))))))
+
+    (testing "failing"
+      (let [db (new-db @client)
+            _  (mc/create db "test")]
+        (is (= 0 (try (mg/with-implicit-transaction
+                        {:client @client}
+                        (fn []
+                          (mc/insert-one db "test" {:a 1})
+                          (mc/insert-one db "test" {nil 2})))
+                      (catch Exception _ (mc/count-documents db "test" {}))))))))
 
   (deftest ^:integration test-with-transaction
     (testing "passing"
       (let [db (new-db @client)
-            _ (mc/create db "test")]
+            _  (mc/create db "test")]
         (with-open [session (mg/start-session @client)]
           (is (= 2 (mg/with-transaction session
-                     (fn []
-                       (mc/insert-one db "test" {:a 1} {:session session})
-                       (mc/insert-one db "test" {:a 2} {:session session})
+                                        (fn []
+                                          (mc/insert-one db "test" {:a 1} {:session session})
+                                          (mc/insert-one db "test" {:a 2} {:session session})
                        (mc/count-documents db "test" {} {:session session}))))))))
 
     (testing "failing"

test/mongo_driver_3/collection_test.clj

@@ -144,7 +144,11 @@
       (is (instance? FindIterable (mc/find db "test" {} {:raw? true}))))
 
     (testing "keywordize"
-      (is (= [{"id" 1}] (mc/find db "test" {} {:keywordize? false :projection {:_id 0 :id 1} :limit 1}))))))
+      (is (= [{"id" 1}] (mc/find db "test" {} {:keywordize? false :projection {:_id 0 :id 1} :limit 1}))))
+
+    (testing "realise-fn"
+      (is (seq? (mc/find db "test" {})))
+      (is (vector? (mc/find db "test" {} {:realise-fn (partial into [])}))))))
 
 (deftest ^:integration test-find-one
   (let [db (new-db @client)
@@ -216,7 +220,13 @@
     (let [db (new-db @client)]
 
       (is (nil? (dissoc (mc/find-one-and-update db "test" {:id 1} {:$set {:r 1}} {:return-new? true}) :_id)))
-      (is (= {:id 1 :r 1} (dissoc (mc/find-one-and-update db "test" {:id 1} {:$set {:r 1}} {:return-new? true :upsert? true}) :_id))))))
+      (is (= {:id 1 :r 1} (dissoc (mc/find-one-and-update db "test" {:id 1} {:$set {:r 1}} {:return-new? true :upsert? true}) :_id)))))
+
+  (testing "aggregation pipeline"
+    (let [db (new-db @client)
+          _ (mc/insert-many db "test" [{:id 1 :a [1 2] :b []} {:id 2 :a [7 8] :b []}])]
+
+      (is (= {:id 1 :a [] :b [1 2]} (dissoc (mc/find-one-and-update db "test" {:id 1} [{:$set {:b :$a}} {:$set {:a []}}] {:return-new? true}) :_id))))))
 
 (deftest ^:integration test-find-one-and-replace
   (testing "return new"
@@ -343,12 +353,7 @@
   (testing "not existing"
     (let [db (new-db @client)
           _ (mc/create db "my-coll")]
-      (is (true? (coll-exists? db "my-coll")))))
-
-  (testing "existing"
-    (let [db (new-db @client)
-          _ (mc/create db "my-coll")]
-      (is (thrown? Exception (mc/create db "my-coll"))))))
+      (is (true? (coll-exists? db "my-coll"))))))
 
 (deftest ^:integration test-rename
   (testing "not existing"
@@ -381,7 +386,11 @@
 (deftest ^:integration test-list-indexes
   (let [db (new-db @client)
         _ (mc/create db "test")]
-    (is (= 1 (count (mc/list-indexes db "test"))) "has default index")))
+    (is (= 1 (count (mc/list-indexes db "test"))) "has default index")
+
+    (testing "realise-fn"
+      (is (seq? (mc/list-indexes db "test")))
+      (is (vector? (mc/list-indexes db "test" {:realise-fn (partial into [])}))))))
 
 (deftest ^:integration test-create-index
   (let [db (new-db @client)

test/mongo_driver_3/model_test.clj

@@ -212,7 +212,7 @@
   (testing "replace one"
     (is (instance? ReplaceOneModel (m/write-model [:replace-one {:filter {:a "b"} :replacement {:a "c"}}])))
     (are [expected arg]
-         (= expected (.isUpsert (.getOptions (m/write-model [:replace-one {:filter {:a "b"} :replacement {:a "c"} :upsert? arg}]))))
+         (= expected (.isUpsert (.getReplaceOptions (m/write-model [:replace-one {:filter {:a "b"} :replacement {:a "c"} :upsert? arg}]))))
       true true
       false false
       false nil))
@@ -220,7 +220,7 @@
   (testing "update many"
     (is (instance? UpdateManyModel (m/write-model [:update-many {:filter {:a "b"} :update {"$set" {:a "c"}}}])))
     (are [expected arg]
-         (= expected (.isUpsert (.getOptions (m/write-model [:update-many {:filter {:a "b"} :update {"$set" {:a "c"}} :upsert? arg}]))))
+         (= expected (.isUpsert (.getOptions ^UpdateManyModel (m/write-model [:update-many {:filter {:a "b"} :update {"$set" {:a "c"}} :upsert? arg}]))))
       true true
       false false
       false nil))